git.m6w6.name Git - awesomized/libmemcached/blob - clients/ms_conn.h

   1 /*
   2  * File:   ms_conn.h
   3  * Author: Mingqiang Zhuang
   4  *
   5  * Created on February 10, 2009
   6  *
   7  * (c) Copyright 2009, Schooner Information Technology, Inc.
   8  * http://www.schoonerinfotech.com/
   9  *
  10  */
  11 #ifndef MS_CONN_H
  12 #define MS_CONN_H
  13
  14 #include <sys/socket.h>
  15 #include <netinet/in.h>
  16 #include <event.h>
  17 #include <netdb.h>
  18
  19 #include "ms_task.h"
  20 #include <libmemcached/memcached/protocol_binary.h>
  21
  22 #ifdef __cplusplus
  23 extern "C" {
  24 #endif
  25
  26 #define DATA_BUFFER_SIZE             (1024 * 1024 + 2048) /* read buffer, 1M + 2k, enough for the max value(1M) */
  27 #define WRITE_BUFFER_SIZE            (32 * 1024)          /* write buffer, 32k */
  28 #define UDP_DATA_BUFFER_SIZE         (1 * 1024 * 1024)    /* read buffer for UDP, 1M */
  29 #define UDP_MAX_PAYLOAD_SIZE         1400                 /* server limit UDP payload size */
  30 #define UDP_MAX_SEND_PAYLOAD_SIZE    1400                 /* mtu size is 1500 */
  31 #define UDP_HEADER_SIZE              8                    /* UDP header size */
  32 #define MAX_SENDBUF_SIZE             (256 * 1024 * 1024)  /* Maximum socket buffer size */
  33 #define SOCK_WAIT_TIMEOUT            10                   /* maximum waiting time of UDP, 10s */
  34 #define EVENT_TIMEOUT                10                   /* maximum waiting time of event,10s */
  35 #define MAX_UDP_PACKET               (1 << 16)            /* maximum UDP packets, 65536 */
  36
  37 /* Initial size of the sendmsg() scatter/gather array. */
  38 #define IOV_LIST_INITIAL             400
  39
  40 /* Initial number of sendmsg() argument structures to allocate. */
  41 #define MSG_LIST_INITIAL             10
  42
  43 /* High water marks for buffer shrinking */
  44 #define READ_BUFFER_HIGHWAT          (2 * DATA_BUFFER_SIZE)
  45 #define UDP_DATA_BUFFER_HIGHWAT      (4 * UDP_DATA_BUFFER_SIZE)
  46 #define IOV_LIST_HIGHWAT             600
  47 #define MSG_LIST_HIGHWAT             100
  48
  49 /* parse udp header */
  50 #define HEADER_TO_REQID(ptr)      ((uint16_t)*ptr * 256 \
  51                                    + (uint16_t)*(ptr + 1))
  52 #define HEADER_TO_SEQNUM(ptr)     ((uint16_t)*(ptr        \
  53                                                + 2) * 256 \
  54                                    + (uint16_t)*(ptr + 3))
  55 #define HEADER_TO_PACKETS(ptr)    ((uint16_t)*(ptr        \
  56                                                + 4) * 256 \
  57                                    + (uint16_t)*(ptr + 5))
  58
  59 /* states of connection */
  60 enum conn_states
  61 {
  62   conn_read,         /* reading in a command line */
  63   conn_write,        /* writing out a simple response */
  64   conn_closing,      /* closing this connection */
  65 };
  66
  67 /* returned states of memcached command */
  68 enum mcd_ret
  69 {
  70   MCD_SUCCESS,                      /* command success */
  71   MCD_FAILURE,                      /* command failure */
  72   MCD_UNKNOWN_READ_FAILURE,         /* unknown read failure */
  73   MCD_PROTOCOL_ERROR,               /* protocol error */
  74   MCD_CLIENT_ERROR,                 /* client error, wrong command */
  75   MCD_SERVER_ERROR,                 /* server error, server run command failed */
  76   MCD_DATA_EXISTS,                  /* object is existent in server */
  77   MCD_NOTSTORED,                    /* server doesn't set the object successfully */
  78   MCD_STORED,                       /* server set the object successfully */
  79   MCD_NOTFOUND,                     /* server not find the object */
  80   MCD_END,                          /* end of the response of get command */
  81   MCD_DELETED,                      /* server delete the object successfully */
  82   MCD_STAT,                         /* response of stats command */
  83 };
  84
  85 /* used to store the current or previous running command state */
  86 typedef struct cmdstat
  87 {
  88   int cmd;                  /* command name */
  89   int retstat;              /* return state of this command */
  90   bool isfinish;            /* if it read all the response data */
  91   uint64_t key_prefix;      /* key prefix */
  92 } ms_cmdstat_t;
  93
  94 /* udp packet structure */
  95 typedef struct udppkt
  96 {
  97   uint8_t *header;          /* udp header of the packet */
  98   char *data;               /* udp data of the packet */
  99   int rbytes;               /* number of data in the packet */
 100   int copybytes;            /* number of copied data in the packet */
 101 } ms_udppkt_t;
 102
 103 /* three protocols supported */
 104 enum protocol
 105 {
 106   ascii_prot = 3,           /* ASCII protocol */
 107   ascii_udp_prot,           /* ASCII UDP protocol*/
 108   binary_prot,              /* binary protocol */
 109 };
 110
 111 /**
 112  *  concurrency structure
 113  *
 114  *  Each thread has a libevent to manage the events of network.
 115  *  Each thread has one or more self-governed concurrencies;
 116  *  each concurrency has one or more socket connections. This
 117  *  concurrency structure includes all the private variables of
 118  *  the concurrency.
 119  */
 120 typedef struct conn
 121 {
 122   int conn_idx;             /* connection index in the thread */
 123   int sfd;                  /* current tcp sock handler of the connection structure */
 124   int udpsfd;               /* current udp sock handler of the connection structure*/
 125   int state;                /* state of the connection */
 126   struct event event;       /* event for libevent */
 127   short ev_flags;           /* event flag for libevent */
 128   short which;              /* which events were just triggered */
 129   bool change_sfd;          /* whether change sfd */
 130
 131   int *tcpsfd;              /* TCP sock array */
 132   int total_sfds;           /* how many socks in the tcpsfd array */
 133   int alive_sfds;           /* alive socks */
 134   int cur_idx;              /* current sock index in tcpsfd array */
 135
 136   ms_cmdstat_t precmd;      /* previous command state */
 137   ms_cmdstat_t currcmd;     /* current command state */
 138
 139   char *rbuf;               /* buffer to read commands into */
 140   char *rcurr;              /* but if we parsed some already, this is where we stopped */
 141   int rsize;                /* total allocated size of rbuf */
 142   int rbytes;               /* how much data, starting from rcur, do we have unparsed */
 143
 144   bool readval;             /* read value state, read known data size */
 145   int rvbytes;              /* total value size need to read */
 146
 147   char *wbuf;               /* buffer to write commands out */
 148   char *wcurr;              /* for multi-get, where we stopped */
 149   int wsize;                /* total allocated size of wbuf */
 150   bool ctnwrite;            /* continue to write */
 151
 152   /* data for the mwrite state */
 153   struct iovec *iov;
 154   int iovsize;              /* number of elements allocated in iov[] */
 155   int iovused;              /* number of elements used in iov[] */
 156
 157   struct msghdr *msglist;
 158   int msgsize;              /* number of elements allocated in msglist[] */
 159   int msgused;              /* number of elements used in msglist[] */
 160   int msgcurr;              /* element in msglist[] being transmitted now */
 161   int msgbytes;             /* number of bytes in current msg */
 162
 163   /* data for UDP clients */
 164   int udp;                          /* is this is a UDP "connection" */
 165   uint32_t request_id;                   /* UDP request ID of current operation, if this is a UDP "connection" */
 166   uint8_t *hdrbuf;                  /* udp packet headers */
 167   int hdrsize;                      /* number of headers' worth of space is allocated */
 168   struct  sockaddr srv_recv_addr;   /* Sent the most recent request to which server */
 169   socklen_t srv_recv_addr_size;
 170
 171   /* udp read buffer */
 172   char *rudpbuf;                    /* buffer to read commands into for udp */
 173   int rudpsize;                     /* total allocated size of rudpbuf */
 174   int rudpbytes;                    /* how much data, starting from rudpbuf */
 175
 176   /* order udp packet */
 177   ms_udppkt_t *udppkt;              /* the offset of udp packet in rudpbuf */
 178   int packets;                      /* number of total packets need to read */
 179   int recvpkt;                      /* number of received packets */
 180   int pktcurr;                      /* current packet in rudpbuf being ordered */
 181   int ordcurr;                      /* current ordered packet */
 182
 183   ms_task_item_t *item_win;         /* task sequence */
 184   int win_size;                     /* current task window size */
 185   uint64_t set_cursor;              /* current set item index in the item window */
 186   ms_task_t curr_task;              /* current running task */
 187   ms_mlget_task_t mlget_task;       /* multi-get task */
 188
 189   int warmup_num;                   /* to run how many warm up operations*/
 190   int remain_warmup_num;            /* left how many warm up operations to run */
 191   int64_t exec_num;                 /* to run how many task operations */
 192   int64_t remain_exec_num;          /* how many remained task operations to run */
 193
 194   /* response time statistic and time out control */
 195   struct timeval start_time;        /* start time of current operation(s) */
 196   struct timeval end_time;          /* end time of current operation(s) */
 197
 198   /* Binary protocol stuff */
 199   protocol_binary_response_header binary_header;    /* local temporary binary header */
 200   enum protocol protocol;                           /* which protocol this connection speaks */
 201 } ms_conn_t;
 202
 203 /* used to generate the key prefix */
 204 uint64_t ms_get_key_prefix(void);
 205
 206
 207 /**
 208  * setup a connection, each connection structure of each
 209  * thread must call this function to initialize.
 210  */
 211 int ms_setup_conn(ms_conn_t *c);
 212
 213
 214 /* after one operation completes, reset the connection */
 215 void ms_reset_conn(ms_conn_t *c, bool timeout);
 216
 217
 218 /**
 219  *  reconnect several disconnected socks in the connection
 220  *  structure, the ever-1-second timer of the thread will check
 221  *  whether some socks in the connections disconnect. if
 222  *  disconnect, reconnect the sock.
 223  */
 224 int ms_reconn_socks(ms_conn_t *c);
 225
 226
 227 /* used to send set command to server */
 228 int ms_mcd_set(ms_conn_t *c, ms_task_item_t *item);
 229
 230
 231 /* used to send the get command to server */
 232 int ms_mcd_get(ms_conn_t *c, ms_task_item_t *item, bool verify);
 233
 234
 235 /* used to send the multi-get command to server */
 236 int ms_mcd_mlget(ms_conn_t *c);
 237
 238
 239 #ifdef __cplusplus
 240 }
 241 #endif
 242
 243 #endif /* end of MS_CONN_H */