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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
/*
* Copyright (C) Max Romanov
* Copyright (C) NGINX, Inc.
*/
#include <nxt_port_hash.h>
// Explicitly using 32 bit types to avoid possible alignment.
typedef struct {
int32_t pid;
uint32_t port_id;
} nxt_pid_port_id_t;
static nxt_int_t
nxt_port_hash_test(nxt_lvlhsh_query_t *lhq, void *data)
{
nxt_port_t *port;
nxt_pid_port_id_t *pid_port_id;
port = data;
pid_port_id = (nxt_pid_port_id_t *) lhq->key.start;
if (lhq->key.length == sizeof(nxt_pid_port_id_t)
&& pid_port_id->pid == port->pid
&& pid_port_id->port_id == port->id)
{
return NXT_OK;
}
return NXT_DECLINED;
}
static const nxt_lvlhsh_proto_t lvlhsh_ports_proto nxt_aligned(64) = {
NXT_LVLHSH_DEFAULT,
nxt_port_hash_test,
nxt_lvlhsh_alloc,
nxt_lvlhsh_free,
};
nxt_port_t *
nxt_port_hash_first(nxt_lvlhsh_t *port_hash, nxt_lvlhsh_each_t *lhe)
{
nxt_memzero(lhe, sizeof(nxt_lvlhsh_each_t));
lhe->proto = &lvlhsh_ports_proto;
return nxt_port_hash_next(port_hash, lhe);
}
void
nxt_port_hash_add(nxt_lvlhsh_t *port_hash, nxt_mp_t *mem_pool,
nxt_port_t *port)
{
nxt_pid_port_id_t pid_port;
nxt_lvlhsh_query_t lhq;
pid_port.pid = port->pid;
pid_port.port_id = port->id;
lhq.key_hash = nxt_murmur_hash2(&pid_port, sizeof(pid_port));
lhq.key.length = sizeof(pid_port);
lhq.key.start = (u_char *) &pid_port;
lhq.proto = &lvlhsh_ports_proto;
lhq.replace = 0;
lhq.value = port;
lhq.pool = mem_pool;
switch (nxt_lvlhsh_insert(port_hash, &lhq)) {
case NXT_OK:
break;
default:
nxt_thread_log_error(NXT_LOG_WARN, "port #%d for pid %PI add failed",
port->id, port->pid);
break;
}
}
void
nxt_port_hash_remove(nxt_lvlhsh_t *port_hash, nxt_mp_t *mem_pool,
nxt_port_t *port)
{
nxt_pid_port_id_t pid_port;
nxt_lvlhsh_query_t lhq;
pid_port.pid = port->pid;
pid_port.port_id = port->id;
lhq.key_hash = nxt_murmur_hash2(&pid_port, sizeof(pid_port));
lhq.key.length = sizeof(pid_port);
lhq.key.start = (u_char *) &pid_port;
lhq.proto = &lvlhsh_ports_proto;
lhq.replace = 0;
lhq.value = port;
lhq.pool = mem_pool;
switch (nxt_lvlhsh_delete(port_hash, &lhq)) {
case NXT_OK:
break;
default:
break;
}
}
nxt_port_t *
nxt_port_hash_find(nxt_lvlhsh_t *port_hash, nxt_pid_t pid,
nxt_port_id_t port_id)
{
nxt_pid_port_id_t pid_port;
nxt_lvlhsh_query_t lhq;
pid_port.pid = pid;
pid_port.port_id = port_id;
lhq.key_hash = nxt_murmur_hash2(&pid_port, sizeof(pid_port));
lhq.key.length = sizeof(pid_port);
lhq.key.start = (u_char *) &pid_port;
lhq.proto = &lvlhsh_ports_proto;
/* TODO lock ports */
if (nxt_lvlhsh_find(port_hash, &lhq) == NXT_OK) {
nxt_thread_log_debug("process port (%PI, %d) found", pid, port_id);
return lhq.value;
}
nxt_thread_log_debug("process port (%PI, %d) not found", pid, port_id);
return NULL;
}
|
