Modules: make unloading module more safe

As we know if a module exports module-side data types,
unload it is not allowed. This rule is the same with
blocked clients in module, because we use background
threads to implement module blocked clients, and it's
not safe to unload a module if there are background
threads running. So it's necessary to check if any
blocked clients running in this module when unload it.

Moreover, after that we can ensure that if no modules,
then no module blocked clients even module unloaded.
So, we can call moduleHandleBlockedClients only when
we have installed modules.

src/module.c

@@ -63,6 +63,7 @@ struct RedisModule {
     int in_call;    /* RM_Call() nesting level */
     int in_hook;    /* Hooks callback nesting level for this module (0 or 1). */
     int options;    /* Module options and capabilities. */
+    int blocked_clients;         /* Count of RedisModuleBlockedClient in this module. */
     RedisModuleInfoFunc info_cb; /* Callback for module to add INFO fields. */
 };
 typedef struct RedisModule RedisModule;
@@ -3961,6 +3962,7 @@ RedisModuleBlockedClient *RM_BlockClient(RedisModuleCtx *ctx, RedisModuleCmdFunc
 
     c->bpop.module_blocked_handle = zmalloc(sizeof(RedisModuleBlockedClient));
     RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
+    ctx->module->blocked_clients++;
 
     /* We need to handle the invalid operation of calling modules blocking
      * commands from Lua or MULTI. We actually create an already aborted
@@ -4119,6 +4121,7 @@ void moduleHandleBlockedClients(void) {
         /* Free 'bc' only after unblocking the client, since it is
          * referenced in the client blocking context, and must be valid
          * when calling unblockClient(). */
+        bc->module->blocked_clients--;
         zfree(bc);
 
         /* Lock again before to iterate the loop. */
@@ -6089,6 +6092,7 @@ int moduleLoad(const char *path, void **module_argv, int module_argc) {
 
     /* Redis module loaded! Register it. */
     dictAdd(modules,ctx.module->name,ctx.module);
+    ctx.module->blocked_clients = 0;
     ctx.module->handle = handle;
     serverLog(LL_NOTICE,"Module '%s' loaded from %s",ctx.module->name,path);
     moduleFreeContext(&ctx);
@@ -6114,6 +6118,9 @@ int moduleUnload(sds name) {
     } else if (listLength(module->usedby)) {
         errno = EPERM;
         return REDISMODULE_ERR;
+    } else if (module->blocked_clients) {
+        errno = EAGAIN;
+        return REDISMODULE_ERR;
     }
 
     /* Give module a chance to clean up. */
@@ -6279,6 +6286,10 @@ NULL
                 errmsg = "the module exports APIs used by other modules. "
                          "Please unload them first and try again";
                 break;
+            case EAGAIN:
+                errmsg = "the module has blocked clients. "
+                         "Please wait them unblocked and try again";
+                break;
             default:
                 errmsg = "operation not possible.";
                 break;

src/server.c

@@ -2104,7 +2104,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
 
     /* Check if there are clients unblocked by modules that implement
      * blocking commands. */
-    moduleHandleBlockedClients();
+    if (moduleCount()) moduleHandleBlockedClients();
 
     /* Try to process pending commands for clients that were just unblocked. */
     if (listLength(server.unblocked_clients))