Changeset 775

Timestamp:

04/02/08 17:41:43 (19 months ago)

Author:

dsallings

Message:

Syscall optimizations from Trond Norbye.

From Trond:
I have been using the last two days to test the modifications I have done to
the binary protocol, and I have not been able to find any new bugs. From my
testing the binary protocol now use the same amount of system calls as the
textual protocol.

I have tried to use a common state machine, and branch the execution path as
late as possible.

Location:

branches/binary/server

Files:

• memcached.c (modified) (28 diffs)
• memcached.h (modified) (2 diffs)

branches/binary/server/memcached.c

@@ -299 +299 @@
 }
 
-static char *prot_text(enum protocol prot) {
+static const char *prot_text(enum protocol prot) {
     char *rv = "unknown";
     switch(prot) {
@@ -384 +384 @@
         } else if (IS_UDP(prot)) {
             fprintf(stderr, "<%d server listening (udp)\n", sfd);
-        } else if (prot == binary_prot) {
-            fprintf(stderr, "<%d new binary client connection\n", sfd);
-        } else if (prot == ascii_prot) {
-            fprintf(stderr, "<%d new ascii client connection\n", sfd);
         } else if (prot == negotiating_prot) {
-            fprintf(stderr, "<%d new auto-negotiating client connection\n", sfd);
+            fprintf(stderr, "<%d new auto-negotiating client connection\n",
+                    sfd);
         } else {
             fprintf(stderr, "<%d new unknown (%d) client connection\n",
@@ -518 +515 @@
 }
 
-static enum conn_states get_init_state(conn *c) {
-    int rv = 0;
-    assert(c != NULL);
-
-    switch(c->protocol) {
-        case binary_prot:
-            rv = conn_bin_init;
-            break;
-        case negotiating_prot:
-            rv = conn_negotiate;
-            break;
-        default:
-            rv = conn_read;
-    }
-    return rv;
-}
-
-/* Set the given connection to its initial state.  The initial state will vary
- * base don protocol type. */
-static void conn_set_init_state(conn *c) {
-    assert(c != NULL);
-
-    conn_set_state(c, get_init_state(c));
-}
-
 /*
  * Shrinks a connection's buffers if they're too big.  This prevents
@@ -601 +573 @@
 }
 
+/**
+ * Convert a state name to a human readable form.
+ */
+static const char *state_text(enum conn_states state) {
+    const char* const statenames[] = { "conn_listening",
+                                       "conn_new_cmd",
+                                       "conn_waiting",
+                                       "conn_read",
+                                       "conn_parse_cmd",
+                                       "conn_write",
+                                       "conn_nread",
+                                       "conn_swallow",
+                                       "conn_closing",
+                                       "conn_mwrite" };
+    return statenames[state];
+}
+
 /*
  * Sets a connection's current state in the state machine. Any special
@@ -608 +597 @@
 static void conn_set_state(conn *c, enum conn_states state) {
     assert(c != NULL);
+    assert(state >= conn_listening && state < conn_max_state);
 
     if (state != c->state) {
-        if (state == conn_read) {
-            conn_shrink(c);
-            assoc_move_next_bucket();
-        }
+        if (settings.verbose > 2) {
+            fprintf(stderr, "%d: going from %s to %s\n",
+                    c->sfd, state_text(c->state),
+                    state_text(state));
+        }
+
         c->state = state;
     }
@@ -811 +803 @@
             fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str);
         c->noreply = false;
-        conn_set_state(c, conn_read);
+        conn_set_state(c, conn_new_cmd);
         return;
     }
@@ -831 +823 @@
 
     conn_set_state(c, conn_write);
-    c->write_and_go = get_init_state(c);
+    c->write_and_go = conn_new_cmd;
     return;
 }
@@ -909 +901 @@
 
 static void write_bin_error(conn *c, int err, int swallow) {
-    char *errstr = "Unknown error";
+    const char *errstr = "Unknown error";
     switch(err) {
+        case ERR_OUT_OF_MEMORY:
+            errstr = "Out of memory";
+            break;
         case ERR_UNKNOWN_CMD:
             errstr = "Unknown command";
@@ -944 +939 @@
         c->write_and_go = conn_swallow;
     } else {
-        c->write_and_go = conn_bin_init;
+        c->write_and_go = conn_new_cmd;
     }
 }
@@ -955 +950 @@
     }
     conn_set_state(c, conn_mwrite);
-    c->write_and_go = conn_bin_init;
+    c->write_and_go = conn_new_cmd;
 }
 
@@ -1113 +1108 @@
     } else {
         if(c->cmd == CMD_GETQ) {
-            conn_set_state(c, conn_bin_init);
+            conn_set_state(c, conn_new_cmd);
         } else {
             write_bin_error(c, ERR_NOT_FOUND, 0);
@@ -1279 +1274 @@
 static void complete_nread_binary(conn *c) {
     assert(c != NULL);
-
-    if(c->cmd < 0) {
-        /* No command defined.  Figure out what they're trying to say. */
-        int i = 0;
-        /* I did a bit of hard-coding around the packet sizes */
-        assert(BIN_PKT_HDR_WORDS == 4);
-        for(i = 0; i<BIN_PKT_HDR_WORDS; i++) {
-            c->bin_header[i] = ntohl(c->bin_header[i]);
-        }
-        if(settings.verbose) {
-            fprintf(stderr, "Read binary protocol data:  %08x %08x %08x %08x\n",
-                c->bin_header[0], c->bin_header[1], c->bin_header[2],
-                c->bin_header[3]);
-        }
-        if((c->bin_header[0] >> 24) != BIN_REQ_MAGIC) {
-            if(settings.verbose) {
-                fprintf(stderr, "Invalid magic:  %x\n", c->bin_header[0] >> 24);
-            }
-            conn_set_state(c, conn_closing);
-            return;
-        }
-
-        c->msgcurr = 0;
-        c->msgused = 0;
-        c->iovused = 0;
-        if (add_msghdr(c) != 0) {
-            out_string(c, "SERVER_ERROR out of memory");
-            return;
-        }
-
-        c->cmd = (c->bin_header[0] >> 16) & 0xff;
-        c->keylen = c->bin_header[0] & 0xffff;
-        c->opaque = c->bin_header[3];
-        if(settings.verbose > 1) {
-            fprintf(stderr,
-                "Command: %d, opaque=%08x, keylen=%d, total_len=%d\n", c->cmd,
-                c->opaque, c->keylen, c->bin_header[2]);
-        }
-        dispatch_bin_command(c);
-    } else {
-        switch(c->substate) {
-            case bin_reading_set_header:
-                process_bin_update(c);
-                break;
-            case bin_read_set_value:
-                complete_update_bin(c);
-                break;
-            case bin_reading_get_key:
-                process_bin_get(c);
-                break;
-            case bin_reading_del_header:
-                process_bin_delete(c);
-                break;
-            case bin_reading_incr_header:
-                complete_incr_bin(c);
-                break;
-            default:
-                fprintf(stderr, "Not handling substate %d\n", c->substate);
-                assert(0);
-        }
-    }
-}
-
-static void reinit_bin_connection(conn *c) {
-    if (settings.verbose > 1)
-        fprintf(stderr, "*** Reinitializing binary connection.\n");
-    c->rlbytes = MIN_BIN_PKT_LENGTH;
-    c->write_and_go = conn_bin_init;
+    assert(c->cmd >= 0);
+
+    switch(c->substate) {
+    case bin_reading_set_header:
+        process_bin_update(c);
+        break;
+    case bin_read_set_value:
+        complete_update_bin(c);
+        break;
+    case bin_reading_get_key:
+        process_bin_get(c);
+        break;
+    case bin_reading_del_header:
+        process_bin_delete(c);
+        break;
+    case bin_reading_incr_header:
+        complete_incr_bin(c);
+        break;
+    default:
+        fprintf(stderr, "Not handling substate %d\n", c->substate);
+        assert(0);
+    }
+}
+
+static void reset_cmd_handler(conn *c) {
     c->cmd = -1;
     c->substate = bin_no_state;
-    c->rbytes = c->wbytes = 0;
-    c->ritem = (char*)c->bin_header;
-    c->rcurr = c->rbuf;
-    c->wcurr = c->wbuf;
     conn_shrink(c);
-    conn_set_state(c, conn_nread);
-}
-
-/* These do the initial protocol switch.  At this point, we should've read
- * exactly one byte, and must treat that byte as the beginning of a command. */
-static void setup_bin_protocol(conn *c) {
-    char *loc = (char*)c->bin_header;
-    if (settings.verbose > 1)
-        fprintf(stderr, "Negotiated protocol as binary.\n");
-
-    c->protocol = binary_prot;
-    reinit_bin_connection(c);
-    /* Emulate a read of the first byte */
-    c->ritem[0] = c->rbuf[0];
-    c->ritem++;
-    c->rlbytes--;
-}
-
-static void setup_ascii_protocol(conn *c) {
-    if (settings.verbose > 1)
-        fprintf(stderr, "Negotiated protocol as ascii.\n");
-    c->protocol = ascii_prot;
-
-    /* We've already got the first letter of the command, so pretend like we
-     * Did a single byte read from try_read_command */
-    c->rcurr = c->rbuf;
-    c->rbytes = 1;
-    conn_set_state(c, conn_read);
+    if (c->rbytes > 0) {
+        conn_set_state(c, conn_parse_cmd);
+    } else {
+        conn_set_state(c, conn_waiting);
+    }
 }
 
 static void complete_nread(conn *c) {
     assert(c != NULL);
+    assert(c->protocol == ascii_prot || c->protocol == binary_prot);
 
     if(c->protocol == ascii_prot) {
         complete_nread_ascii(c);
-    } else if(c->protocol == binary_prot) {
+    } else if (c->protocol == binary_prot) {
         complete_nread_binary(c);
-    } else if(c->protocol == negotiating_prot) {
-        /* The first byte is either BIN_REQ_MAGIC, or we're speaking ascii */
-        if ((c->rbuf[0] & 0xff) == BIN_REQ_MAGIC)
-            setup_bin_protocol(c);
-        else
-            setup_ascii_protocol(c);
-    } else {
-        assert(0); /* XXX:  Invalid case.  Should probably do more here. */
     }
 }
@@ -1579 +1497 @@
         c->wbytes = bytes;
         conn_set_state(c, conn_write);
-        c->write_and_go = get_init_state(c);
+        c->write_and_go = conn_new_cmd;
     } else {
         out_string(c, "SERVER_ERROR out of memory writing stats");
@@ -1866 +1784 @@
                  */
 
-                if(return_cas == true)
+                if (return_cas)
                 {
                   /* Goofy mid-flight realloc. */
@@ -2362 +2280 @@
                 c->gen = gen;
             }
-            conn_set_init_state(c);
+            conn_set_state(c, conn_new_cmd);
             return;
         } else {
@@ -2456 +2374 @@
  */
 static int try_read_command(conn *c) {
-    char *el, *cont;
-
     assert(c != NULL);
     assert(c->rcurr <= (c->rbuf + c->rsize));
-
-    if (c->rbytes == 0)
-        return 0;
-    el = memchr(c->rcurr, '\n', c->rbytes);
-    if (!el)
-        return 0;
-    cont = el + 1;
-    if ((el - c->rcurr) > 1 && *(el - 1) == '\r') {
-        el--;
-    }
-    *el = '\0';
-
-    assert(cont <= (c->rcurr + c->rbytes));
-
-    process_command(c, c->rcurr);
-
-    c->rbytes -= (cont - c->rcurr);
-    c->rcurr = cont;
-
-    assert(c->rcurr <= (c->rbuf + c->rsize));
+    assert(c->rbytes > 0);
+
+    if (c->protocol == negotiating_prot)  {
+        if ((c->rbuf[0] & 0xff) == BIN_REQ_MAGIC) {
+            c->protocol = binary_prot;
+        } else {
+            c->protocol = ascii_prot;
+        }
+
+        if (settings.verbose) {
+            fprintf(stderr, "%d: Client using the %s protocol\n", c->sfd,
+                    prot_text(c->protocol));
+        }
+    }
+
+    if (c->protocol == binary_prot) {
+        /* Do we have the complete packet header? */
+        if (c->rbytes < MIN_BIN_PKT_LENGTH) {
+            /* need more data! */
+            return 0;
+        } else {
+            int i = 0;
+            memcpy(c->bin_header, c->rcurr, sizeof(c->bin_header));
+            assert(BIN_PKT_HDR_WORDS == 4);
+            for (i = 0; i<BIN_PKT_HDR_WORDS; i++) {
+                c->bin_header[i] = ntohl(c->bin_header[i]);
+            }
+
+            if (settings.verbose) {
+                fprintf(stderr,
+                        "Read binary protocol data:  %08x %08x %08x %08x\n",
+                        c->bin_header[0], c->bin_header[1], c->bin_header[2],
+                        c->bin_header[3]);
+            }
+
+            if ((c->bin_header[0] >> 24) != BIN_REQ_MAGIC) {
+                if (settings.verbose) {
+                    fprintf(stderr, "Invalid magic:  %x\n",
+                            c->bin_header[0] >> 24);
+                }
+                conn_set_state(c, conn_closing);
+                return 0;
+            }
+
+            c->msgcurr = 0;
+            c->msgused = 0;
+            c->iovused = 0;
+            if (add_msghdr(c) != 0) {
+                out_string(c, "SERVER_ERROR out of memory");
+                return 0;
+            }
+
+            c->cmd = (c->bin_header[0] >> 16) & 0xff;
+            c->keylen = c->bin_header[0] & 0xffff;
+            c->opaque = c->bin_header[3];
+            if (settings.verbose > 1) {
+                fprintf(stderr,
+                        "Command: %d, opaque=%08x, keylen=%d, total_len=%d\n",
+                        c->cmd, c->opaque, c->keylen, c->bin_header[2]);
+            }
+
+            dispatch_bin_command(c);
+
+            c->rbytes -= MIN_BIN_PKT_LENGTH;
+            c->rcurr += MIN_BIN_PKT_LENGTH;
+        }
+    } else {
+        char *el, *cont;
+
+        if (c->rbytes == 0)
+            return 0;
+        el = memchr(c->rcurr, '\n', c->rbytes);
+        if (!el)
+            return 0;
+        cont = el + 1;
+        if ((el - c->rcurr) > 1 && *(el - 1) == '\r') {
+            el--;
+        }
+        *el = '\0';
+
+        assert(cont <= (c->rcurr + c->rbytes));
+
+        process_command(c, c->rcurr);
+
+        c->rbytes -= (cont - c->rcurr);
+        c->rcurr = cont;
+
+        assert(c->rcurr <= (c->rbuf + c->rsize));
+    }
 
     return 1;
@@ -2527 +2512 @@
  * before reading, move the remaining incomplete fragment of a command
  * (if any) to the beginning of the buffer.
- * return 0 if there's nothing to read on the first read.
+ * @return 1 data received
+ *         0 no data received
+ *        -1 an error occured (on the socket) (or client closed connection)
+ *        -2 memory error (failed to allocate more memory)
  */
 static int try_read_network(conn *c) {
@@ -2550 +2538 @@
                 out_string(c, "SERVER_ERROR out of memory reading request");
                 c->write_and_go = conn_closing;
-                return 1;
+                return -2;
             }
             c->rcurr = c->rbuf = new_rbuf;
@@ -2571 +2559 @@
         }
         if (res == 0) {
-            /* connection closed */
-            conn_set_state(c, conn_closing);
-            return 1;
+            return -1;
         }
         if (res == -1) {
-            if (errno == EAGAIN || errno == EWOULDBLOCK) break;
-            /* Should close on unhandled errors. */
-            conn_set_state(c, conn_closing);
-            return 1;
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                break;
+            }
+            return -1;
         }
     }
@@ -2694 +2680 @@
     bool stop = false;
     int sfd, flags = 1;
-    enum conn_states init_state; /* initial state for a new connection */
     socklen_t addrlen;
     struct sockaddr_storage addr;
@@ -2727 +2712 @@
                 break;
             }
-            init_state = get_init_state(c);
-
-            dispatch_conn_new(sfd, init_state, EV_READ | EV_PERSIST,
+
+            dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,
                                      DATA_BUFFER_SIZE, c->protocol);
-
-            break;
-
-        case conn_negotiate:
-            if (settings.verbose > 1)
-                fprintf(stderr, "Negotiating protocol for a new connection\n");
-            c->rlbytes = 1;
-            c->ritem = c->rbuf;
-            c->rcurr = c->rbuf;
-            c->wcurr = c->wbuf;
-            conn_set_state(c, conn_nread);
-            break;
-
-        case conn_read:
-            if (try_read_command(c) != 0) {
-                continue;
-            }
-            if ((IS_UDP(c->protocol) ? try_read_udp(c) : try_read_network(c)) != 0) {
-                continue;
-            }
-            /* we have no command line and no data to read from network */
+            stop = true;
+            break;
+
+        case conn_waiting:
             if (!update_event(c, EV_READ | EV_PERSIST)) {
                 if (settings.verbose > 0)
@@ -2758 +2725 @@
                 break;
             }
+
+            conn_set_state(c, conn_read);
             stop = true;
             break;
 
-        case conn_bin_init: /* Reinitialize a binary connection */
-            reinit_bin_connection(c);
+        case conn_read:
+            res = IS_UDP(c->protocol) ? try_read_udp(c) : try_read_network(c);
+            switch (res) {
+            case 0 :
+                conn_set_state(c, conn_waiting);
+                break;
+            case 1:
+                conn_set_state(c, conn_parse_cmd);
+                break;
+            case -1:
+                conn_set_state(c, conn_closing);
+                break;
+            case -2: /* Failed to allocate more memory */
+                /* State already set by try_read_network */
+                break;
+            }
+            break;
+
+        case conn_parse_cmd :
+            if (try_read_command(c) == 0) {
+                /* wee need more data! */
+                conn_set_state(c, conn_waiting);
+            }
+
+            break;
+
+        case conn_new_cmd:
+            reset_cmd_handler(c);
+            assoc_move_next_bucket();
             break;
 
@@ -2773 +2769 @@
             if (c->rbytes > 0) {
                 int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes;
-                memcpy(c->ritem, c->rcurr, tocopy);
+                memmove(c->ritem, c->rcurr, tocopy);
                 c->ritem += tocopy;
                 c->rlbytes -= tocopy;
                 c->rcurr += tocopy;
                 c->rbytes -= tocopy;
-                break;
+                if (c->rlbytes == 0) {
+                    break;
+                }
             }
 
@@ -2814 +2812 @@
             /* we are reading sbytes and throwing them away */
             if (c->sbytes == 0) {
-                conn_set_init_state(c);
+                conn_set_state(c, conn_new_cmd);
                 break;
             }
@@ -2898 +2896 @@
                         conn_set_state(c, c->write_and_go);
                     } else {
-                        conn_set_init_state(c);
+                        conn_set_state(c, conn_new_cmd);
                     }
                 } else if (c->state == conn_write) {
@@ -3286 +3284 @@
            "-v            verbose (print errors/warnings while in event loop)\n"
            "-vv           very verbose (also print client commands/reponses)\n"
+           "-vvv          extremely verbose (also print internal state transitions)\n"
            "-h            print this help and exit\n"
            "-i            print memcached and libevent license\n"
@@ -3481 +3480 @@
 
     /* process arguments */
-    while ((c = getopt(argc, argv, "a:bp:B:s:U:m:Mc:khirvdl:u:P:f:s:n:t:D:")) != -1) {
+    while ((c = getopt(argc, argv, "a:bp:s:U:m:Mc:khirvdl:u:P:f:s:n:t:D:L")) != -1) {
         switch (c) {
         case 'a':
@@ -3565 +3564 @@
             settings.detail_enabled = 1;
             break;
+        case 'L' :
 #if defined(HAVE_GETPAGESIZES) && defined(HAVE_MEMCNTL)
-        case 'L' :
             if (enable_large_pages() == 0) {
                 preallocate = true;
             }
-            break;
 #endif
+            break;
         default:
             fprintf(stderr, "Illegal argument \"%c\"\n", c);

branches/binary/server/memcached.h

@@ -176 +176 @@
 #define ITEM_ntotal(item) (sizeof(struct _stritem) + (item)->nkey + 1 + (item)->nsuffix + (item)->nbytes)
 
+/**
+ * NOTE: If you modify this table you _MUST_ update the function state_text
+ */
 enum conn_states {
     conn_listening,  /** the socket which listens for connections */
+    conn_new_cmd,    /** Prepare connection for next command */
+    conn_waiting,    /** waiting for a readable socket */
     conn_read,       /** reading in a command line */
+    conn_parse_cmd,  /** try to parse a command from the input buffer */
     conn_write,      /** writing out a simple response */
     conn_nread,      /** reading in a fixed number of bytes */
@@ -184 +190 @@
     conn_closing,    /** closing this connection */
     conn_mwrite,     /** writing out many items sequentially */
-    conn_bin_init,   /** Reinitializing a binary protocol connection */
-    conn_negotiate,  /** Negotiating a protocol */
+    conn_max_state,  /** Max state value (used for assertion) */
 };