MEDIUM: pipes: Never allocate more than maxpipes

Commit c81d794822cf8fa3c2e163cbf4cba80ad0dd9a47 made it so we could have
one pipe pool per thread group, however it did not limit how many pipes
we'd allocate per thread group, so one thread group could end up having
all the pipes. Prevent that by allowing a maximum of maxpipes / nbtgroup
per thread group.

diff --git a/src/pipe.c b/src/pipe.c
index 2a1dd92f7cc39b84167ee9ecc057f8d571e69074..9e1fa26df393734e87a400a07402ddb760b4f8e5 100644 (file)
--- a/src/pipe.c
+++ b/src/pipe.c
@@ -32,9 +32,46 @@ DECLARE_STATIC_TYPED_POOL(pool_head_pipe, "pipe", struct pipe);
  */
 static struct {
        struct pipe *live;        /* pipes which are still ready to use */
+       uint max;                 /* max pipes this pool may own */
+       uint count;               /* pipes owned by this pool (used + free) */
        __decl_thread(HA_SPINLOCK_T lock); /* lock used to protect the list */
 } pipe_pools[MAX_TGROUPS] __attribute__((aligned(64)));
 
+/* returns the index of the pool the current thread must use */
+static inline int pipe_pool_id(void)
+{
+       return (global.tune.options & GTUNE_NO_TG_FD_SHARING) ? tgid - 1 : 0;
+}
+
+/* returns the number of threads sharing the current thread's pool */
+static inline int pipe_pool_threads(void)
+{
+       if (global.tune.options & GTUNE_NO_TG_FD_SHARING)
+               return tg->count;
+       return global.nbthread;
+}
+
+static int init_pipes_per_thread(void)
+{
+       uint pools, id, base, rem;
+
+       if (global.tune.options & GTUNE_NO_TG_FD_SHARING) {
+               pools = global.nbtgroups;
+               id    = tgid - 1;
+       } else {
+               pools = 1;
+               id    = 0;
+       }
+
+       base = global.maxpipes / pools;
+       rem  = global.maxpipes % pools;
+
+       /* all the threads of a pool compute the same value */
+       HA_ATOMIC_STORE(&pipe_pools[id].max, base + (id < rem));
+       return 1;
+}
+REGISTER_PER_THREAD_INIT(init_pipes_per_thread);
+
 static THREAD_LOCAL int local_pipes_free = 0;  /* #cache objects   */
 static THREAD_LOCAL struct pipe *local_pipes = NULL;
 
@@ -49,6 +86,7 @@ struct pipe *get_pipe()
        struct pipe *ret = NULL;
        int pipefd[2];
        int grp;
+       uint count;
 
        ret = local_pipes;
        if (likely(ret)) {
@@ -59,7 +97,7 @@ struct pipe *get_pipe()
                goto out;
        }
 
-       grp = (global.tune.options & GTUNE_NO_TG_FD_SHARING) ? tgid - 1 : 0;
+       grp = pipe_pool_id();
 
        if (likely(pipe_pools[grp].live)) {
                HA_SPIN_LOCK(PIPES_LOCK, &pipe_pools[grp].lock);
@@ -74,9 +112,15 @@ struct pipe *get_pipe()
                }
        }
 
+       count = HA_ATOMIC_LOAD(&pipe_pools[grp].count);
+       for (;; __ha_cpu_relax()) {
+               if (count >= HA_ATOMIC_LOAD(&pipe_pools[grp].max))
+                       return NULL;
+               if (HA_ATOMIC_CAS(&pipe_pools[grp].count, &count, count + 1))
+                       break;
+       }
+
        HA_ATOMIC_INC(&pipes_used);
-       if (pipes_used + pipes_free >= global.maxpipes)
-               goto fail;
 
        ret = pool_alloc(pool_head_pipe);
        if (!ret)
@@ -98,6 +142,7 @@ struct pipe *get_pipe()
  fail:
        pool_free(pool_head_pipe, ret);
        HA_ATOMIC_DEC(&pipes_used);
+       HA_ATOMIC_DEC(&pipe_pools[grp].count);
        return NULL;
 
 }
@@ -111,6 +156,7 @@ void kill_pipe(struct pipe *p)
        close(p->cons);
        pool_free(pool_head_pipe, p);
        HA_ATOMIC_DEC(&pipes_used);
+       HA_ATOMIC_DEC(&pipe_pools[pipe_pool_id()].count);
 }
 
 /* put back a unused pipe into the live pool. If it still has data in it, it is
@@ -126,15 +172,15 @@ void put_pipe(struct pipe *p)
                return;
        }
 
-       if (likely(local_pipes_free * global.nbthread < global.maxpipes - pipes_used)) {
+       grp = pipe_pool_id();
+
+       if (likely((uint)(local_pipes_free + 1) * pipe_pool_threads() <= HA_ATOMIC_LOAD(&pipe_pools[grp].max))) {
                p->next = local_pipes;
                local_pipes = p;
                local_pipes_free++;
                goto out;
        }
 
-       grp = (global.tune.options & GTUNE_NO_TG_FD_SHARING) ? tgid - 1 : 0;
-
        HA_SPIN_LOCK(PIPES_LOCK, &pipe_pools[grp].lock);
        p->next = pipe_pools[grp].live;
        pipe_pools[grp].live = p;