shm

Connection.h

@@ -13,6 +13,7 @@
 #include <event2/event.h>
 #include <event2/util.h>
 
+#include "bipbuffer.h"
 #include "AdaptiveSampler.h"
 #include "cmdline.h"
 #include "ConnectionOptions.h"
@@ -534,6 +535,8 @@ public:
   void event_callback2(short events);
   void read_callback1();
   void read_callback2();
+  void read_callback1_v1();
+  void read_callback2_v1();
   // event callbacks
   void write_callback();
   void timer_callback();
@@ -623,6 +626,8 @@ private:
   unsigned char* buffer_read[MAX_LEVELS];
   unsigned char* buffer_write_pos[MAX_LEVELS];
   unsigned char* buffer_read_pos[MAX_LEVELS];
+  unsigned char* buffer_lasthdr[MAX_LEVELS];
+  unsigned char* buffer_leftover[MAX_LEVELS];
   uint32_t buffer_read_n[MAX_LEVELS];
   uint32_t buffer_write_n[MAX_LEVELS];
   uint32_t buffer_read_nbytes[MAX_LEVELS];
@@ -654,6 +659,7 @@ private:
   void issue_sasl();
   int issue_op(Operation* op);
   int issue_noop(int level = 1);
+  size_t fill_read_buffer(int level, int *extra);
   int issue_touch(const char* key, int valuelen, double now, int level);
   int issue_delete(const char* key, double now, uint32_t flags);
   int issue_get_with_len(const char* key, int valuelen, double now, bool quiet, uint32_t flags, Operation *l1 = NULL);
@@ -675,4 +681,161 @@ private:
   bool consume_resp_line(evbuffer *input, bool &done);
 };
 
+class ConnectionMultiApproxShm {
+public:
+  ConnectionMultiApproxShm(options_t options, bool sampling = true);
+
+  ~ConnectionMultiApproxShm();
+
+  int do_connect();
+
+  double start_time; // Time when this connection began operations.
+  ConnectionStats stats;
+  options_t options;
+
+  bool is_ready() { return read_state == IDLE; }
+  void set_priority(int pri);
+
+  void start_loading();
+  void reset();
+  bool check_exit_condition(double now = 0.0);
+
+  void event_callback1(short events);
+  void event_callback2(short events);
+  void read_callback1();
+  void read_callback2();
+  void read_callback1_v1();
+  void read_callback2_v1();
+  // event callbacks
+  void write_callback();
+  void timer_callback();
+  
+  int eof;
+  uint32_t get_cid();
+  //void set_queue(ConcurrentQueue<string> *a_trace_queue);
+  int  add_to_wb_keys(string wb_key);
+  int  add_to_copy_keys(string key);
+  int  add_to_touch_keys(string key);
+  void del_wb_keys(string wb_key);
+  void del_copy_keys(string key);
+  void del_touch_keys(string key);
+  void set_g_wbkeys(unordered_map<string,vector<Operation*>> *a_wb_keys);
+  void set_queue(queue<Operation*> *a_trace_queue);
+  void set_lock(pthread_mutex_t* a_lock);
+  int send_write_buffer(int level);
+  int add_get_op_to_queue(Operation *pop, int level);
+  int add_set_to_queue(Operation *pop, int level, const char *value);
+  size_t handle_response_batch(unsigned char *rbuf_pos, resp_t *resp, 
+                                    size_t read_bytes, size_t consumed_bytes,
+                                    int level, int extra);
+  void drive_write_machine_shm(double now = 0.0);
+  bipbuf_t* bipbuf_in[3];
+  bipbuf_t* bipbuf_out[3];
+  pthread_mutex_t* lock_in[3];
+  pthread_mutex_t* lock_out[3];
+
+private:
+  string hostname1;
+  string hostname2;
+  string port;
+
+  double o_percent;
+  int trace_queue_n;
+
+  struct event *timer; // Used to control inter-transmission time.
+  double next_time;    // Inter-transmission time parameters.
+  double last_rx;      // Used to moderate transmission rate.
+  double last_tx;
+
+  enum read_state_enum {
+    INIT_READ,
+    CONN_SETUP,
+    LOADING,
+    IDLE,
+    WAITING_FOR_GET,
+    WAITING_FOR_SET,
+    WAITING_FOR_DELETE,
+    MAX_READ_STATE,
+  };
+
+  enum write_state_enum {
+    INIT_WRITE,
+    ISSUING,
+    WAITING_FOR_TIME,
+    WAITING_FOR_OPQ,
+    MAX_WRITE_STATE,
+  };
+
+  read_state_enum read_state;
+  write_state_enum write_state;
+
+  // Parameters to track progress of the data loader.
+  int loader_issued, loader_completed;
+
+  uint32_t *opaque;
+  int *issue_buf_size;
+  int *issue_buf_n;
+  unsigned char **issue_buf_pos;
+  unsigned char **issue_buf;
+  bool last_quiet1;
+  bool last_quiet2;
+  uint32_t total;
+  uint32_t cid;
+  uint32_t gets;
+  uint32_t gloc;
+  uint32_t ghits;
+  uint32_t sloc;
+  uint32_t esets;
+  uint32_t isets;
+  uint32_t iloc;
+
+
+  //std::vector<std::queue<Operation>> op_queue;
+  Operation ***op_queue;
+  uint32_t *op_queue_size;
+
+
+  Generator *valuesize;
+  Generator *keysize;
+  KeyGenerator *keygen;
+  Generator *iagen;
+  pthread_mutex_t* lock;
+  unordered_map<string,vector<Operation*>> *g_wb_keys;
+  queue<Operation*> *trace_queue;
+
+  // state machine functions / event processing
+  void pop_op(Operation *op);
+  void output_op(Operation *op, int type, bool was_found);
+  //void finish_op(Operation *op);
+  void finish_op(Operation *op,int was_hit);
+  int issue_getsetorset(double now = 0.0);
+
+  // request functions
+  void issue_sasl();
+  int issue_op(Operation* op);
+  void issue_noop(int level = 1);
+  size_t fill_read_buffer(int level, int *extra);
+  int issue_touch(const char* key, int valuelen, double now, int level);
+  int issue_delete(const char* key, double now, uint32_t flags);
+  int issue_get_with_len(const char* key, int valuelen, double now, bool quiet, uint32_t flags, Operation *l1 = NULL);
+  int issue_get_with_len(Operation *pop, double now, bool quiet, uint32_t flags, Operation *l1 = NULL);
+  int issue_set(const char* key, const char* value, int length, double now, uint32_t flags);
+  int issue_set(Operation *pop, const char* value, double now, uint32_t flags);
+
+  int read_response_l1(); 
+  void read_response_l2();
+  // protocol fucntions
+  int set_request_ascii(const char* key, const char* value, int length);
+  int set_request_binary(const char* key, const char* value, int length);
+  int set_request_resp(const char* key, const char* value, int length);
+  
+  int get_request_ascii(const char* key);
+  int get_request_binary(const char* key);
+  int get_request_resp(const char* key);
+
+  bool consume_binary_response(evbuffer *input);
+  bool consume_ascii_line(evbuffer *input, bool &done);
+  bool consume_resp_line(evbuffer *input, bool &done);
+};
+
 #endif

ConnectionMultiApprox.cc

@@ -640,10 +640,9 @@ int ConnectionMultiApprox::issue_getsetorset(double now) {
             //
             
             Operation *Op = trace_queue->front(); 
-            trace_queue->pop();
             //Operation *Op = g_trace_queue.dequeue(); 
             
-            if (Op->type == Operation::SASL) {
+            if (Op == NULL || trace_queue->size() <= 0 || Op->type == Operation::SASL) {
                 eof = 1;
                 cid_rate.insert( {cid, 100 } );
                 fprintf(stderr,"cid %d done\n",cid);
@@ -664,6 +663,7 @@ int ConnectionMultiApprox::issue_getsetorset(double now) {
                 return 1;
             } 
             
+            trace_queue->pop();
 
             
             //trace_queue->pop();
@@ -906,11 +906,11 @@ int ConnectionMultiApprox::issue_touch(const char* key, int valuelen, double now
   pop->valuelen = valuelen;
   pop->type = Operation::TOUCH;
   pop->opaque = opaque[level]++;
-  pop->flags = flags;
   op_queue[level][pop->opaque] = pop;
-  //op_queue[level].push(op);
   op_queue_size[level]++;
   
+  pop->flags = flags;
+  
   if (opaque[level] > OPAQUE_MAX) {
       opaque[level] = 1;
   }

ConnectionOptions.h

@@ -7,6 +7,7 @@ typedef struct {
   int apps;
   int rand_admit;
   bool ratelimit;
+  bool v1callback;
   int threshold;
   int wb_all;
   bool miss_through;

SConstruct

@@ -44,8 +44,8 @@ if not conf.CheckFunc('pthread_barrier_init'):
 
 env = conf.Finish()
 
-env.Append(CFLAGS = ' -O0 -Wall -g --std=c++17 -lstdc++fs')
-env.Append(CPPFLAGS = ' -O0 -Wall -g --std=c++17 -lstdc++fs')
+env.Append(CFLAGS = '-O3 -Wall -g --std=c++17 -lstdc++fs')
+env.Append(CPPFLAGS = '-O3 -Wall -g --std=c++17 -lstdc++fs')
 #env.Append(CFLAGS = ' -O3 -Wall -g --std=c++17 -lstdc++fs')
 #env.Append(CPPFLAGS = ' -O3 -Wall -g --std=c++17 -lstdc++fs')
 #env.Append(CFLAGS = ' -O3 -Wall -g')
@@ -62,7 +62,7 @@ env.Append(CPPFLAGS = ' -O0 -Wall -g --std=c++17 -lstdc++fs')
 env.Command(['cmdline.cc', 'cmdline.h'], 'cmdline.ggo', 'gengetopt < $SOURCE')
 
 src = Split("""mutilate.cc cmdline.cc log.cc distributions.cc util.cc
-               Connection.cc ConnectionMulti.cc ConnectionMultiApprox.cc Protocol.cc Generator.cc""")
+               Connection.cc ConnectionMulti.cc ConnectionMultiApprox.cc ConnectionMultiApproxBatch.cc ConnectionMultiApproxShm.cc Protocol.cc Generator.cc bipbuffer.cc""")
 
 if not env['HAVE_POSIX_BARRIER']: # USE_POSIX_BARRIER:
     src += ['barrier.cc']

binary_protocol.h

@@ -1,9 +1,12 @@
 #ifndef BINARY_PROTOCOL_H
 #define	BINARY_PROTOCOL_H
 
+#include <stdint.h>
+
 #define CMD_GET  0x00
 #define CMD_GETQ 0x09
 #define CMD_TOUCH 0x1c
+#define CMD_TOUCH 0x1c
 #define CMD_DELETE 0x04
 #define CMD_SET  0x01
 #define CMD_NOOP 0x0a

bipbuffer.cc

@@ -0,0 +1,180 @@
+/**
+ * Copyright (c) 2011, Willem-Hendrik Thiart
+ * Use of this source code is governed by a BSD-style license that can be
+ * found in the LICENSE.bipbuffer file.
+ *
+ * @file
+ * @author  Willem Thiart himself@willemthiart.com
+ */
+
+//#include "stdio.h"
+#include <stdlib.h>
+
+/* for memcpy */
+#include <string.h>
+
+#include "bipbuffer.h"
+
+static size_t bipbuf_sizeof(const unsigned int size)
+{
+    return sizeof(bipbuf_t) + size;
+}
+
+int bipbuf_unused(const bipbuf_t* me)
+{
+    if (1 == me->b_inuse)
+        /* distance between region B and region A */
+        return me->a_start - me->b_end;
+    else
+        return me->size - me->a_end;
+}
+
+int bipbuf_size(const bipbuf_t* me)
+{
+    return me->size;
+}
+
+int bipbuf_used(const bipbuf_t* me)
+{
+    return (me->a_end - me->a_start) + me->b_end;
+}
+
+void bipbuf_init(bipbuf_t* me, const unsigned int size)
+{
+    me->a_start = me->a_end = me->b_end = 0;
+    me->size = size;
+    me->b_inuse = 0;
+}
+
+bipbuf_t *bipbuf_new(const unsigned int size)
+{
+    bipbuf_t *me = (bipbuf_t*)malloc(bipbuf_sizeof(size));
+    if (!me)
+        return NULL;
+    bipbuf_init(me, size);
+    return me;
+}
+
+void bipbuf_free(bipbuf_t* me)
+{
+    free(me);
+}
+
+int bipbuf_is_empty(const bipbuf_t* me)
+{
+    return me->a_start == me->a_end;
+}
+
+/* find out if we should turn on region B
+ * ie. is the distance from A to buffer's end less than B to A? */
+static void __check_for_switch_to_b(bipbuf_t* me)
+{
+    if (me->size - me->a_end < me->a_start - me->b_end)
+        me->b_inuse = 1;
+}
+
+/* TODO: DOCUMENT THESE TWO FUNCTIONS */
+unsigned char *bipbuf_request(bipbuf_t* me, const int size)
+{
+    if (bipbuf_unused(me) < size)
+        return 0;
+    if (1 == me->b_inuse)
+    {
+        return (unsigned char *)me->data + me->b_end;
+    }
+    else
+    {
+        return (unsigned char *)me->data + me->a_end;
+    }
+}
+
+int bipbuf_push(bipbuf_t* me, const int size)
+{
+    if (bipbuf_unused(me) < size)
+        return 0;
+
+    if (1 == me->b_inuse)
+    {
+        me->b_end += size;
+    }
+    else
+    {
+        me->a_end += size;
+    }
+
+    __check_for_switch_to_b(me);
+    return size;
+}
+
+int bipbuf_offer(bipbuf_t* me, const unsigned char *data, const int size)
+{
+    /* not enough space */
+    if (bipbuf_unused(me) < size)
+        return 0;
+
+    if (1 == me->b_inuse)
+    {
+        memcpy(me->data + me->b_end, data, size);
+        me->b_end += size;
+    }
+    else
+    {
+        memcpy(me->data + me->a_end, data, size);
+        me->a_end += size;
+    }
+
+    __check_for_switch_to_b(me);
+    return size;
+}
+
+unsigned char *bipbuf_peek(const bipbuf_t* me, const unsigned int size)
+{
+    /* make sure we can actually peek at this data */
+    if (me->size < me->a_start + size)
+        return NULL;
+
+    if (bipbuf_is_empty(me))
+        return NULL;
+
+    return (unsigned char *)me->data + me->a_start;
+}
+
+unsigned char *bipbuf_peek_all(const bipbuf_t* me, unsigned int *size)
+{
+    if (bipbuf_is_empty(me))
+        return NULL;
+
+    *size = me->a_end - me->a_start;
+    return (unsigned char*)me->data + me->a_start;
+}
+
+unsigned char *bipbuf_poll(bipbuf_t* me, const unsigned int size)
+{
+    if (bipbuf_is_empty(me))
+        return NULL;
+
+    /* make sure we can actually poll this data */
+    if (me->size < me->a_start + size)
+        return NULL;
+
+    void *end = me->data + me->a_start;
+    me->a_start += size;
+
+    /* we seem to be empty.. */
+    if (me->a_start == me->a_end)
+    {
+        /* replace a with region b */
+        if (1 == me->b_inuse)
+        {
+            me->a_start = 0;
+            me->a_end = me->b_end;
+            me->b_end = me->b_inuse = 0;
+        }
+        else
+            /* safely move cursor back to the start because we are empty */
+            me->a_start = me->a_end = 0;
+    }
+
+    __check_for_switch_to_b(me);
+    return (unsigned char*) end;
+}

bipbuffer.h

@@ -0,0 +1,92 @@
+#ifndef BIPBUFFER_H
+#define BIPBUFFER_H
+
+#define BIPBUFSIZE 4*1024*1024
+#include "binary_protocol.h"
+
+extern "C" {
+    typedef struct
+    {
+        unsigned long int size;
+    
+        /* region A */
+        unsigned int a_start, a_end;
+    
+        /* region B */
+        unsigned int b_end;
+    
+        /* is B inuse? */
+        int b_inuse;
+    
+        unsigned char data[BIPBUFSIZE];
+    } bipbuf_t;
+
+/**
+ * Create a new bip buffer.
+ *
+ * malloc()s space
+ *
+ * @param[in] size The size of the buffer */
+bipbuf_t *bipbuf_new(const unsigned int size);
+
+/**
+ * Initialise a bip buffer. Use memory provided by user.
+ *
+ * No malloc()s are performed.
+ *
+ * @param[in] size The size of the array */
+void bipbuf_init(bipbuf_t* me, const unsigned int size);
+
+/**
+ * Free the bip buffer */
+void bipbuf_free(bipbuf_t *me);
+
+/* TODO: DOCUMENTATION */
+unsigned char *bipbuf_request(bipbuf_t* me, const int size);
+int bipbuf_push(bipbuf_t* me, const int size);
+
+/**
+ * @param[in] data The data to be offered to the buffer
+ * @param[in] size The size of the data to be offered
+ * @return number of bytes offered */
+int bipbuf_offer(bipbuf_t *me, const unsigned char *data, const int size);
+
+/**
+ * Look at data. Don't move cursor
+ *
+ * @param[in] len The length of the data to be peeked
+ * @return data on success, NULL if we can't peek at this much data */
+unsigned char *bipbuf_peek(const bipbuf_t* me, const unsigned int len);
+
+/**
+ * Look at data. Don't move cursor
+ *
+ * @param[in] len The length of the data returned
+ * @return data on success, NULL if nothing available */
+unsigned char *bipbuf_peek_all(const bipbuf_t* me, unsigned int *len);
+
+/**
+ * Get pointer to data to read. Move the cursor on.
+ *
+ * @param[in] len The length of the data to be polled
+ * @return pointer to data, NULL if we can't poll this much data */
+unsigned char *bipbuf_poll(bipbuf_t* me, const unsigned int size);
+
+/**
+ * @return the size of the bipbuffer */
+int bipbuf_size(const bipbuf_t* me);
+
+/**
+ * @return 1 if buffer is empty; 0 otherwise */
+int bipbuf_is_empty(const bipbuf_t* me);
+
+/**
+ * @return how much space we have assigned */
+int bipbuf_used(const bipbuf_t* cb);
+
+/**
+ * @return bytes of unused space */
+int bipbuf_unused(const bipbuf_t* me);
+
+}
+#endif /* BIPBUFFER_H */

cmdline.ggo

@@ -10,7 +10,9 @@ option "quiet" - "Disable log messages."
 
 text "\nBasic options:"
 
+option "use_shm" - "use shared memory"
 option "ratelimit" - "limit conns from exceeding each other in requests"
+option "v1callback" - "use v1 readcallbacks"
 option "server" s "Memcached server hostname[:port].  \
 Repeat to specify multiple servers." string multiple
 option "unix_socket" - "Use UNIX socket instead of TCP."

mutilate.cc

@@ -1629,7 +1629,7 @@ void do_mutilate(const vector<string>& servers, options_t& options,
     event_config_free(config);
     evdns_base_free(evdns, 0);
     event_base_free(base);
-  } else if (servers.size() == 2 && ! ( args.approx_given || args.approx_batch_given)) {
+  } else if (servers.size() == 2 && ! ( args.approx_given || args.approx_batch_given || args.use_shm_given)) {
     vector<ConnectionMulti*> connections;
     vector<ConnectionMulti*> server_lead;
 
@@ -2079,7 +2079,12 @@ void do_mutilate(const vector<string>& servers, options_t& options,
         }
       }
       if (restart) continue;
-      else break;
+      else {
+          for (ConnectionMultiApproxBatch *conn: connections) {
+              fprintf(stderr,"tid %ld, cid: %d\n",pthread_self(),conn->get_cid());
+          }
+          break;
+      }
 
     }
 
@@ -2101,6 +2106,61 @@ void do_mutilate(const vector<string>& servers, options_t& options,
     event_config_free(config);
     evdns_base_free(evdns, 0);
     event_base_free(base);
+  } else if (servers.size() == 2 && args.use_shm_given) {
+    vector<ConnectionMultiApproxShm*> connections;
+
+    int conns = args.measure_connections_given ? args.measure_connections_arg :
+      options.connections;
+
+    srand(time(NULL));
+    for (int c = 0; c < conns; c++) {
+
+
+      ConnectionMultiApproxShm* conn = new ConnectionMultiApproxShm(options,args.agentmode_given ? false : true);
+      int connected = 0;
+      if (conn && conn->do_connect()) {
+          connected = 1;
+      }
+      int cid = conn->get_cid();
+      
+      if (connected) {
+        fprintf(stderr,"cid %d gets trace_queue\nfirst: %s\n",cid,trace_queue->at(cid)->front()->key);
+        if (g_lock != NULL) {
+            conn->set_g_wbkeys(g_wb_keys);
+            conn->set_lock(g_lock);
+        }
+        conn->set_queue(trace_queue->at(cid));
+        connections.push_back(conn);
+      } else {
+        fprintf(stderr,"conn multi: %d, not connected!!\n",c);
+
+      }
+    }
+    
+    // wait for all threads to reach here
+    pthread_barrier_wait(&barrier);
+    //fprintf(stderr,"Start = %f\n", start);
+    double start = get_time();
+    double now = start;
+    for (ConnectionMultiApproxShm *conn: connections) {
+        conn->start_time = now;
+        conn->drive_write_machine_shm(now);
+    }
+
+
+
+    if (master && !args.scan_given && !args.search_given)
+      V("stopped at %f  options.time = %d", get_time(), options.time);
+
+    // Tear-down and accumulate stats.
+    for (ConnectionMultiApproxShm *conn: connections) {
+      stats.accumulate(conn->stats);
+      delete conn;
+    }
+
+    stats.start = start;
+    stats.stop = now;
+
   }
 }
 
@@ -2117,6 +2177,7 @@ void args_to_options(options_t* options) {
   options->threshold = args.threshold_arg;
   options->wb_all = args.wb_all_arg;
   options->ratelimit = args.ratelimit_given;
+  options->v1callback = args.v1callback_given;
   if (args.inclusives_given) {
     memset(options->inclusives,0,256);
     strncpy(options->inclusives,args.inclusives_arg,256);