driver: release the queues of half-initialized pipes

lib/driver.c

@@ -387,10 +387,9 @@ _log_dest_driver_unregister_counters(LogDestDriver *self)
   stats_unlock();
 }
 
-gboolean
-log_dest_driver_deinit_method(LogPipe *s)
+static inline void
+_log_dest_driver_release_queues(LogDestDriver *self)
 {
-  LogDestDriver *self = (LogDestDriver *) s;
   GList *l, *l_next;
 
   for (l = self->queues; l; l = l_next)
@@ -404,7 +403,16 @@ log_dest_driver_deinit_method(LogPipe *s)
        * which automatically frees the ref on the list too */
       log_dest_driver_release_queue(self, log_queue_ref(q));
     }
+
   g_assert(self->queues == NULL);
+}
+
+gboolean
+log_dest_driver_deinit_method(LogPipe *s)
+{
+  LogDestDriver *self = (LogDestDriver *) s;
+
+  _log_dest_driver_release_queues(self);
 
   _log_dest_driver_unregister_counters(self);
 
@@ -431,12 +439,9 @@ void
 log_dest_driver_free(LogPipe *s)
 {
   LogDestDriver *self = (LogDestDriver *) s;
-  GList *l;
 
-  for (l = self->queues; l; l = l->next)
-    {
-      log_queue_unref((LogQueue *) l->data);
-    }
-  g_list_free(self->queues);
+  /* half-initialized pipes can't release their queue in deinit() */
+  _log_dest_driver_release_queues(self);
+
   log_driver_free(s);
 }

lib/logthrdest/logthrdestdrv.c

@@ -1352,7 +1352,11 @@ _create_workers(LogThreadedDestDriver *self, gint stats_level, StatsClusterKeyBu
 
       self->workers[self->created_workers] = dw;
       if (!_acquire_worker_queue(dw, stats_level, driver_sck_builder))
-        return FALSE;
+        {
+          /* failed worker needs to be destroyed */
+          self->created_workers++;
+          return FALSE;
+        }
     }
 
   return TRUE;

news/bugfix-128.md

@@ -0,0 +1,3 @@
+`disk-buffer()`: fix crash when pipeline initialization fails
+
+`log_queue_disk_free_method: assertion failed: (!qdisk_started(self->qdisk))`