[PATCH 1/3] virthreadpool: Allow setting identity for workers

[Daniel P. Berrangé]

On Tue, Oct 26, 2021 at 12:17:06PM +0200, Michal Privoznik wrote:
> In some cases the worker func running inside the pool may rely on
> virIdentity. While worker func could check for identity and set
> one it is not optimal - it may not have access to the identity of
> the thread creating the pool and thus would have to call
> virIdentityGetSystem(). Allow passing identity when creating the
> pool.

I wonder if we should have an identity passed via virThreadPoolSendJob,
so whatever queues the job can preserve its identity ?

> 
> Signed-off-by: Michal Privoznik <mprivozn at redhat.com>
> ---
>  src/nwfilter/nwfilter_dhcpsnoop.c |  1 +
>  src/qemu/qemu_driver.c            |  4 +++-
>  src/rpc/virnetserver.c            |  1 +
>  src/util/virthreadpool.c          | 12 ++++++++++++
>  src/util/virthreadpool.h          | 12 +++++++-----
>  5 files changed, 24 insertions(+), 6 deletions(-)
> 
> diff --git a/src/nwfilter/nwfilter_dhcpsnoop.c b/src/nwfilter/nwfilter_dhcpsnoop.c
> index 4b62a7b661..b0297f0741 100644
> --- a/src/nwfilter/nwfilter_dhcpsnoop.c
> +++ b/src/nwfilter/nwfilter_dhcpsnoop.c
> @@ -1333,6 +1333,7 @@ virNWFilterDHCPSnoopThread(void *req0)
>          worker = virThreadPoolNewFull(1, 1, 0,
>                                        virNWFilterDHCPDecodeWorker,
>                                        "dhcp-decode",
> +                                      NULL,
>                                        req);
>      }
>  
> diff --git a/src/qemu/qemu_driver.c b/src/qemu/qemu_driver.c
> index 6997dc7dea..a8bf0ecc6f 100644
> --- a/src/qemu/qemu_driver.c
> +++ b/src/qemu/qemu_driver.c
> @@ -914,7 +914,9 @@ qemuStateInitialize(bool privileged,
>       * running domains since there might occur some QEMU monitor
>       * events that will be dispatched to the worker pool */
>      qemu_driver->workerPool = virThreadPoolNewFull(0, 1, 0, qemuProcessEventHandler,
> -                                                   "qemu-event", qemu_driver);
> +                                                   "qemu-event",
> +                                                   NULL,
> +                                                   qemu_driver);
>      if (!qemu_driver->workerPool)
>          goto error;
>  
> diff --git a/src/rpc/virnetserver.c b/src/rpc/virnetserver.c
> index dc8f32b095..c7b4939398 100644
> --- a/src/rpc/virnetserver.c
> +++ b/src/rpc/virnetserver.c
> @@ -378,6 +378,7 @@ virNetServer *virNetServerNew(const char *name,
>                                                priority_workers,
>                                                virNetServerHandleJob,
>                                                "rpc-worker",
> +                                              NULL,
>                                                srv)))
>          goto error;
>  
> diff --git a/src/util/virthreadpool.c b/src/util/virthreadpool.c
> index 92b7cac286..7bf4333885 100644
> --- a/src/util/virthreadpool.c
> +++ b/src/util/virthreadpool.c
> @@ -55,6 +55,8 @@ struct _virThreadPool {
>      virThreadPoolJobList jobList;
>      size_t jobQueueDepth;
>  
> +    virIdentity *identity;
> +
>      virMutex mutex;
>      virCond cond;
>      virCond quit_cond;
> @@ -99,6 +101,9 @@ static void virThreadPoolWorker(void *opaque)
>  
>      virMutexLock(&pool->mutex);
>  
> +    if (pool->identity)
> +        virIdentitySetCurrent(pool->identity);
> +
>      while (1) {
>          /* In order to support async worker termination, we need ensure that
>           * both busy and free workers know if they need to terminated. Thus,
> @@ -219,6 +224,7 @@ virThreadPoolNewFull(size_t minWorkers,
>                       size_t prioWorkers,
>                       virThreadPoolJobFunc func,
>                       const char *name,
> +                     virIdentity *identity,
>                       void *opaque)
>  {
>      virThreadPool *pool;
> @@ -234,6 +240,9 @@ virThreadPoolNewFull(size_t minWorkers,
>      pool->jobName = name;
>      pool->jobOpaque = opaque;
>  
> +    if (identity)
> +        pool->identity = g_object_ref(identity);
> +
>      if (virMutexInit(&pool->mutex) < 0)
>          goto error;
>      if (virCondInit(&pool->cond) < 0)
> @@ -300,6 +309,9 @@ void virThreadPoolFree(virThreadPool *pool)
>      virMutexLock(&pool->mutex);
>      virThreadPoolDrainLocked(pool);
>  
> +    if (pool->identity)
> +        g_object_unref(pool->identity);
> +
>      g_free(pool->workers);
>      virMutexUnlock(&pool->mutex);
>      virMutexDestroy(&pool->mutex);
> diff --git a/src/util/virthreadpool.h b/src/util/virthreadpool.h
> index 619d128e9a..c6b9f31916 100644
> --- a/src/util/virthreadpool.h
> +++ b/src/util/virthreadpool.h
> @@ -22,17 +22,19 @@
>  #pragma once
>  
>  #include "internal.h"
> +#include "viridentity.h"
>  
>  typedef struct _virThreadPool virThreadPool;
>  
>  typedef void (*virThreadPoolJobFunc)(void *jobdata, void *opaque);
>  
>  virThreadPool *virThreadPoolNewFull(size_t minWorkers,
> -                                      size_t maxWorkers,
> -                                      size_t prioWorkers,
> -                                      virThreadPoolJobFunc func,
> -                                      const char *name,
> -                                      void *opaque) ATTRIBUTE_NONNULL(4);
> +                                    size_t maxWorkers,
> +                                    size_t prioWorkers,
> +                                    virThreadPoolJobFunc func,
> +                                    const char *name,
> +                                    virIdentity *identity,
> +                                    void *opaque) ATTRIBUTE_NONNULL(4);
>  
>  size_t virThreadPoolGetMinWorkers(virThreadPool *pool);
>  size_t virThreadPoolGetMaxWorkers(virThreadPool *pool);
> -- 
> 2.32.0
> 

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|