src/packet_pool.h


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

#pragma once

#include <al/array.h>

#include "util/packet.h"
#include "util/thread/thread.h"

#include "loop.h"

// Steps:
//  packet_get (block)           \  Thread 2
//  packet_submit (signal async) /
//  callback called              \  Thread 1
//  packet_return (unblock)      /
//  -- Close --
//  pool_disable()
//  join(thread1) && join(thread2)
//  pool_free()

enum {
    NNWT_PACKET_POOL_KEEP = 0,
    NNWT_PACKET_POOL_RETURN
};

struct nn_packet_pool {
    struct nn_event_loop *loop;
    u32 size;
    bool grow;
    bool flushing;
    bool disabled;
    struct nn_cond cond;
    struct nn_mutex mutex;
    array(struct nn_packet *) empty;
    array(struct nn_packet *) ready;
    array(struct nn_packet *) sending;
    ev_async signal;
    void (*callback)(void *, struct nn_packet *);
    void *userdata;
};

void nn_packet_pool_init(struct nn_packet_pool *pool, u32 size,
    struct nn_event_loop *loop, void (*callback)(void *, struct nn_packet *), void *userdata);
struct nn_packet *nn_packet_pool_get(struct nn_packet_pool *pool);
void nn_packet_pool_submit(struct nn_packet_pool *pool, struct nn_packet *packet);
void nn_packet_pool_flush(struct nn_packet_pool *pool);
void nn_packet_pool_lock(struct nn_packet_pool *pool);
void nn_packet_pool_return(struct nn_packet_pool *pool, struct nn_packet *packet);
void nn_packet_pool_unlock(struct nn_packet_pool *pool);
void nn_packet_pool_disable(struct nn_packet_pool *pool);
void nn_packet_pool_enable(struct nn_packet_pool *pool);
void nn_packet_pool_free(struct nn_packet_pool *pool);