Merge branch 'moar_perf' into keydbpro_moarperf

Former-commit-id: d0990a465c3c5c29b71f239d84a875af40699281

README.md

@@ -13,7 +13,9 @@ KeyDB maintains full compatibility with the Redis protocol, modules, and scripts
 
 On the same hardware KeyDB can perform twice as many queries per second as Redis, with 60% lower latency. Active-Replication simplifies hot-spare failover allowing you to easily distribute writes over replicas and use simple TCP based load balancing/failover. KeyDB's higher performance allows you to do more on less hardware which reduces operation costs and complexity.
 
-<img src=https://cdn-images-1.medium.com/max/1400/1*s7mTb7Qb0kxc951mz8bdgA.png width=420 height=300/><img src=https://cdn-images-1.medium.com/max/1400/1*R00A5U4AFGohGOYHMfT6fA.png height=300/>
+<img src="https://keydb.dev/assets/img/blog/5x_opspersecVSdatasize.PNG"/>
+
+See the full benchmark results and setup information here: https://docs.keydb.dev/blog/2019/10/07/blog-post/
 
 Why fork Redis?
 ---------------

keydb.conf

@@ -1561,3 +1561,10 @@ server-threads 2
 # replicas will still sync in the normal way and incorrect ordering when
 # bringing up replicas can result in data loss (the first master will win).
 # active-replica yes
+
+# Enable Pro?  KeyDB pro provides support for pro only features
+#   note: you may omit the license key to demo pro features for a limited time
+# enable-pro [License Key]
+
+# Enable FLASH support? (Pro Only)
+# storage-provider flash /path/to/flash/db

src/aelocker.h

@@ -11,9 +11,11 @@ public:
 
     void arm(client *c) // if a client is passed, then the client is already locked
     {
+        if (m_fArmed)
+            return;
+        
         if (c != nullptr)
         {
-            serverAssert(!m_fArmed);
             serverAssert(c->lock.fOwnLock());
 
             if (!aeTryAcquireLock(true /*fWeak*/))    // avoid locking the client if we can

src/aof.cpp

@@ -165,11 +165,15 @@ void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
 
     /* Install a file event to send data to the rewrite child if there is
      * not one already. */
+    if (!g_pserver->aof_rewrite_pending) {
+        g_pserver->aof_rewrite_pending = true;
         aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [] {
+            g_pserver->aof_rewrite_pending = false;
             if (g_pserver->aof_pipe_write_data_to_child >= 0)
                 aeCreateFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, g_pserver->aof_pipe_write_data_to_child, AE_WRITABLE, aofChildWriteDiffData, NULL);
         });
     }
+}
 
 /* Write the buffer (possibly composed of multiple blocks) into the specified
  * fd. If a short write or any other error happens -1 is returned,

src/server.cpp

@@ -3689,9 +3689,6 @@ int processCommand(client *c, int callFlags) {
 
     incrementMvccTstamp();
 
-    if (!locker.isArmed())
-        locker.arm(c);
-
     /* Handle the maxmemory directive.
      *
      * Note that we do not want to reclaim memory if we are here re-entering
@@ -3699,6 +3696,7 @@ int processCommand(client *c, int callFlags) {
      * condition, to avoid mixing the propagation of scripts with the
      * propagation of DELs due to eviction. */
     if (g_pserver->maxmemory && !g_pserver->lua_timedout) {
+        locker.arm(c);
         int out_of_memory = freeMemoryIfNeededAndSafe() == C_ERR;
         /* freeMemoryIfNeeded may flush replica output buffers. This may result
          * into a replica, that may be the active client, to be freed. */
@@ -3718,6 +3716,9 @@ int processCommand(client *c, int callFlags) {
 
     /* Don't accept write commands if there are problems persisting on disk
      * and if this is a master instance. */
+    if (c->cmd->flags & CMD_WRITE || c->cmd->proc == pingCommand)
+    {
+        locker.arm(c);
         int deny_write_type = writeCommandsDeniedByDiskError();
         if (deny_write_type != DISK_ERROR_TYPE_NONE &&
             listLength(g_pserver->masters) == 0 &&
@@ -3757,6 +3758,7 @@ int processCommand(client *c, int callFlags) {
             addReply(c, shared.roslaveerr);
             return C_OK;
         }
+    }
 
     /* Only allow a subset of commands in the context of Pub/Sub if the
      * connection is in RESP2 mode. With RESP3 there are no limits. */
@@ -3770,6 +3772,9 @@ int processCommand(client *c, int callFlags) {
         return C_OK;
     }
 
+    if (listLength(g_pserver->masters))
+    {
+        locker.arm(c);
         /* Only allow commands with flag "t", such as INFO, SLAVEOF and so on,
         * when replica-serve-stale-data is no and we are a replica with a broken
         * link with master. */
@@ -3781,6 +3786,7 @@ int processCommand(client *c, int callFlags) {
             addReply(c, shared.masterdownerr);
             return C_OK;
         }
+    }
 
     /* Loading DB? Return an error if the command has not the
      * CMD_LOADING flag. */
@@ -3814,6 +3820,7 @@ int processCommand(client *c, int callFlags) {
         queueMultiCommand(c);
         addReply(c,shared.queued);
     } else {
+        locker.arm(c);
         call(c,callFlags);
         c->woff = g_pserver->master_repl_offset;
         if (listLength(g_pserver->ready_keys))

src/server.h

@@ -2104,6 +2104,7 @@ struct redisServer {
     int aof_stop_sending_diff;     /* If true stop sending accumulated diffs
                                       to child process. */
     sds aof_child_diff;             /* AOF diff accumulator child side. */
+    int aof_rewrite_pending = 0;    /* is a call to aofChildWriteDiffData already queued? */
     /* RDB persistence */
     long long dirty;                /* Changes to DB from the last save */
     long long dirty_before_bgsave;  /* Used to restore dirty on failed BGSAVE */