git.m6w6.name Git - m6w6/libmemcached/blob - contrib/bin/memaslap/ms_conn.h

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