Document of c-modernization-kit (porter) 1.0.0

Loading...
Searching...
No Matches
potrSendQueue.c
Go to the documentation of this file.
1
13
14#include <stdlib.h>
15#include <string.h>
16
17#ifndef _WIN32
18 #include <time.h>
19#endif /* _WIN32 */
20
21#include <porter_const.h>
22
23#include "potrSendQueue.h"
24
25/* --------------------------------------------------------------------------
26 * プラットフォーム別 ミューテックス・条件変数 ラッパーマクロ
27 * -------------------------------------------------------------------------- */
28#ifndef _WIN32
29 #define POTR_MUTEX_INIT(m) pthread_mutex_init((m), NULL)
30 #define POTR_MUTEX_LOCK(m) pthread_mutex_lock(m)
31 #define POTR_MUTEX_UNLOCK(m) pthread_mutex_unlock(m)
32 #define POTR_MUTEX_DESTROY(m) pthread_mutex_destroy(m)
33 #define POTR_COND_INIT(c) pthread_cond_init((c), NULL)
34 #define POTR_COND_WAIT(c, m) pthread_cond_wait((c), (m))
35 #define POTR_COND_SIGNAL(c) pthread_cond_signal(c)
36 #define POTR_COND_BROADCAST(c) pthread_cond_broadcast(c)
37 #define POTR_COND_DESTROY(c) pthread_cond_destroy(c)
38#else /* _WIN32 */
39 #define POTR_MUTEX_INIT(m) InitializeCriticalSection(m)
40 #define POTR_MUTEX_LOCK(m) EnterCriticalSection(m)
41 #define POTR_MUTEX_UNLOCK(m) LeaveCriticalSection(m)
42 #define POTR_MUTEX_DESTROY(m) DeleteCriticalSection(m)
43 #define POTR_COND_INIT(c) InitializeConditionVariable(c)
44 #define POTR_COND_WAIT(c, m) SleepConditionVariableCS((c), (m), INFINITE)
45 #define POTR_COND_SIGNAL(c) WakeConditionVariable(c)
46 #define POTR_COND_BROADCAST(c) WakeAllConditionVariable(c)
47 #define POTR_COND_DESTROY(c) ((void)0) /* Windows は破棄不要 */
48#endif /* _WIN32 */
49
50/* doxygen コメントは、ヘッダに記載 */
51 int potr_send_queue_init(PotrSendQueue *q, size_t depth, uint16_t max_payload)
52{
53 size_t i;
54
55 memset(q, 0, sizeof(*q));
56
57 q->entries = (PotrPayloadElem *)malloc(depth * sizeof(PotrPayloadElem));
58 q->payload_pool = (uint8_t *)malloc(depth * (size_t)max_payload);
59
60 if (q->entries == NULL || q->payload_pool == NULL)
61 {
62 free(q->entries);
63 free(q->payload_pool);
64 q->entries = NULL;
65 q->payload_pool = NULL;
66 return POTR_ERROR;
67 }
68
69 q->depth = depth;
70
71 for (i = 0; i < depth; i++)
72 {
73 q->entries[i].peer_id = POTR_PEER_NA;
74 q->entries[i].flags = 0;
75 q->entries[i].payload_len = 0;
76 q->entries[i].payload = q->payload_pool + i * (size_t)max_payload;
77 }
78
79 POTR_MUTEX_INIT(&q->mutex);
80 POTR_COND_INIT(&q->not_empty);
81 POTR_COND_INIT(&q->not_full);
82 POTR_COND_INIT(&q->drained);
83 return POTR_SUCCESS;
84}
85
86/* doxygen コメントは、ヘッダに記載 */
87 void potr_send_queue_destroy(PotrSendQueue *q)
88{
89 POTR_COND_DESTROY(&q->drained);
90 POTR_COND_DESTROY(&q->not_full);
91 POTR_COND_DESTROY(&q->not_empty);
92 POTR_MUTEX_DESTROY(&q->mutex);
93 free(q->entries);
94 free(q->payload_pool);
95 q->entries = NULL;
96 q->payload_pool = NULL;
97}
98
99/* doxygen コメントは、ヘッダに記載 */
100 int potr_send_queue_push(PotrSendQueue *q, PotrPeerId peer_id,
101 uint16_t flags,
102 const void *payload, uint16_t payload_len)
103{
104 POTR_MUTEX_LOCK(&q->mutex);
105
106 if (q->count + q->inflight >= q->depth)
107 {
108 POTR_MUTEX_UNLOCK(&q->mutex);
109 return POTR_ERROR;
110 }
111
112 q->entries[q->tail].peer_id = peer_id;
113 q->entries[q->tail].flags = flags;
114 q->entries[q->tail].payload_len = payload_len;
115 memcpy(q->entries[q->tail].payload, payload, payload_len);
116 q->tail = (q->tail + 1U) % q->depth;
117 q->count++;
118
119 POTR_COND_SIGNAL(&q->not_empty);
120 POTR_MUTEX_UNLOCK(&q->mutex);
121
122 return POTR_SUCCESS;
123}
124
125/* doxygen コメントは、ヘッダに記載 */
126 int potr_send_queue_push_wait(PotrSendQueue *q, PotrPeerId peer_id,
127 uint16_t flags,
128 const void *payload, uint16_t payload_len,
129 volatile int *running)
130{
131 POTR_MUTEX_LOCK(&q->mutex);
132
133 /* count + inflight < depth が保証されるまで待機する。
134 inflight エントリもプールスロットを占有するため、count だけでは不足。 */
135 while (q->count + q->inflight >= q->depth)
136 {
137 if (!*running)
138 {
139 POTR_MUTEX_UNLOCK(&q->mutex);
140 return POTR_ERROR;
141 }
142 POTR_COND_WAIT(&q->not_full, &q->mutex);
143 }
144
145 q->entries[q->tail].peer_id = peer_id;
146 q->entries[q->tail].flags = flags;
147 q->entries[q->tail].payload_len = payload_len;
148 memcpy(q->entries[q->tail].payload, payload, payload_len);
149 q->tail = (q->tail + 1U) % q->depth;
150 q->count++;
151
152 POTR_COND_SIGNAL(&q->not_empty);
153 POTR_MUTEX_UNLOCK(&q->mutex);
154
155 return POTR_SUCCESS;
156}
157
158/* doxygen コメントは、ヘッダに記載 */
159 int potr_send_queue_pop(PotrSendQueue *q, PotrPayloadElem *out, volatile int *running)
160{
161 POTR_MUTEX_LOCK(&q->mutex);
162
163 while (q->count == 0)
164 {
165 if (!*running)
166 {
167 POTR_MUTEX_UNLOCK(&q->mutex);
168 return POTR_ERROR;
169 }
170 POTR_COND_WAIT(&q->not_empty, &q->mutex);
171 }
172
173 *out = q->entries[q->head];
174 q->head = (q->head + 1U) % q->depth;
175 q->count--;
176 q->inflight++;
177
178 /* count + inflight は変化しない (count-- と inflight++ が相殺) ため
179 not_full シグナルは complete() が担う */
180 POTR_MUTEX_UNLOCK(&q->mutex);
181 return POTR_SUCCESS;
182}
183
184/* doxygen コメントは、ヘッダに記載 */
185 int potr_send_queue_peek(PotrSendQueue *q, PotrPayloadElem *out)
186{
187 POTR_MUTEX_LOCK(&q->mutex);
188
189 if (q->count == 0)
190 {
191 POTR_MUTEX_UNLOCK(&q->mutex);
192 return POTR_ERROR;
193 }
194
195 *out = q->entries[q->head]; /* head は送信スレッドのみが変更するので安全 */
196
197 POTR_MUTEX_UNLOCK(&q->mutex);
198 return POTR_SUCCESS;
199}
200
201/* doxygen コメントは、ヘッダに記載 */
202 int potr_send_queue_peek_timed(PotrSendQueue *q, PotrPayloadElem *out,
203 uint32_t timeout_ms)
204{
205 POTR_MUTEX_LOCK(&q->mutex);
206
207 if (q->count == 0)
208 {
209#ifndef _WIN32
210 struct timespec abs_ts;
211 clock_gettime(CLOCK_REALTIME, &abs_ts);
212 abs_ts.tv_sec += (time_t)(timeout_ms / 1000U);
213 abs_ts.tv_nsec += (long)((timeout_ms % 1000U) * 1000000UL);
214 if (abs_ts.tv_nsec >= 1000000000L)
215 {
216 abs_ts.tv_sec++;
217 abs_ts.tv_nsec -= 1000000000L;
218 }
219 pthread_cond_timedwait(&q->not_empty, &q->mutex, &abs_ts);
220#else /* _WIN32 */
221 SleepConditionVariableCS(&q->not_empty, &q->mutex, (DWORD)timeout_ms);
222#endif /* _WIN32 */
223 }
224
225 if (q->count == 0)
226 {
227 POTR_MUTEX_UNLOCK(&q->mutex);
228 return POTR_ERROR;
229 }
230
231 *out = q->entries[q->head];
232
233 POTR_MUTEX_UNLOCK(&q->mutex);
234 return POTR_SUCCESS;
235}
236
237/* doxygen コメントは、ヘッダに記載 */
238 int potr_send_queue_try_pop(PotrSendQueue *q, PotrPayloadElem *out)
239{
240 POTR_MUTEX_LOCK(&q->mutex);
241
242 if (q->count == 0)
243 {
244 POTR_MUTEX_UNLOCK(&q->mutex);
245 return POTR_ERROR;
246 }
247
248 *out = q->entries[q->head];
249 q->head = (q->head + 1U) % q->depth;
250 q->count--;
251 q->inflight++;
252
253 POTR_MUTEX_UNLOCK(&q->mutex);
254 return POTR_SUCCESS;
255}
256
257/* doxygen コメントは、ヘッダに記載 */
258 void potr_send_queue_complete(PotrSendQueue *q)
259{
260 POTR_MUTEX_LOCK(&q->mutex);
261
262 if (q->inflight > 0U)
263 {
264 q->inflight--;
265 }
266
267 if (q->count == 0U && q->inflight == 0U)
268 {
269 POTR_COND_BROADCAST(&q->drained);
270 }
271
272 /* inflight 減少により count + inflight < depth となる可能性があるため
273 push_wait で待機中のスレッドを起床させる */
274 POTR_COND_SIGNAL(&q->not_full);
275
276 POTR_MUTEX_UNLOCK(&q->mutex);
277}
278
279/* doxygen コメントは、ヘッダに記載 */
280 void potr_send_queue_wait_drained(PotrSendQueue *q)
281{
282 POTR_MUTEX_LOCK(&q->mutex);
283
284 while (q->count > 0U || q->inflight > 0U)
285 {
286 POTR_COND_WAIT(&q->drained, &q->mutex);
287 }
288
289 POTR_MUTEX_UNLOCK(&q->mutex);
290}
291
292/* doxygen コメントは、ヘッダに記載 */
293 void potr_send_queue_shutdown(PotrSendQueue *q)
294{
295 POTR_MUTEX_LOCK(&q->mutex);
296 POTR_COND_BROADCAST(&q->not_empty);
297 POTR_COND_BROADCAST(&q->not_full);
298 POTR_MUTEX_UNLOCK(&q->mutex);
299}
POTR_PEER_NA
#define POTR_PEER_NA
ピア ID 未割当を示す予約値。 1:1 モードのコールバックで渡される (ピアの概念がない)。 potrSend() に N:1 モードで指定した場合はエラーを返す。
Definition porter_type.h:258
POTR_SUCCESS
#define POTR_SUCCESS
成功の戻り値を表す定数。
Definition porter_const.h:30
POTR_ERROR
#define POTR_ERROR
失敗の戻り値を表す定数。
Definition porter_const.h:31
porter_const.h
通信ライブラリの定数ファイル。
PotrPeerId
uint32_t PotrPeerId
ピア識別子。
Definition porter_type.h:32
POTR_COND_DESTROY
#define POTR_COND_DESTROY(c)
Definition potrSendQueue.c:37
POTR_MUTEX_UNLOCK
#define POTR_MUTEX_UNLOCK(m)
Definition potrSendQueue.c:31
potr_send_queue_init
int potr_send_queue_init(PotrSendQueue *q, size_t depth, uint16_t max_payload)
Definition potrSendQueue.c:51
POTR_MUTEX_LOCK
#define POTR_MUTEX_LOCK(m)
Definition potrSendQueue.c:30
potr_send_queue_push
int potr_send_queue_push(PotrSendQueue *q, PotrPeerId peer_id, uint16_t flags, const void *payload, uint16_t payload_len)
Definition potrSendQueue.c:100
potr_send_queue_complete
void potr_send_queue_complete(PotrSendQueue *q)
Definition potrSendQueue.c:258
potr_send_queue_pop
int potr_send_queue_pop(PotrSendQueue *q, PotrPayloadElem *out, volatile int *running)
Definition potrSendQueue.c:159
potr_send_queue_peek_timed
int potr_send_queue_peek_timed(PotrSendQueue *q, PotrPayloadElem *out, uint32_t timeout_ms)
Definition potrSendQueue.c:202
POTR_COND_SIGNAL
#define POTR_COND_SIGNAL(c)
Definition potrSendQueue.c:35
POTR_COND_INIT
#define POTR_COND_INIT(c)
Definition potrSendQueue.c:33
POTR_MUTEX_INIT
#define POTR_MUTEX_INIT(m)
Definition potrSendQueue.c:29
potr_send_queue_peek
int potr_send_queue_peek(PotrSendQueue *q, PotrPayloadElem *out)
Definition potrSendQueue.c:185
potr_send_queue_shutdown
void potr_send_queue_shutdown(PotrSendQueue *q)
Definition potrSendQueue.c:293
potr_send_queue_try_pop
int potr_send_queue_try_pop(PotrSendQueue *q, PotrPayloadElem *out)
Definition potrSendQueue.c:238
POTR_COND_WAIT
#define POTR_COND_WAIT(c, m)
Definition potrSendQueue.c:34
potr_send_queue_destroy
void potr_send_queue_destroy(PotrSendQueue *q)
Definition potrSendQueue.c:87
POTR_COND_BROADCAST
#define POTR_COND_BROADCAST(c)
Definition potrSendQueue.c:36
potr_send_queue_push_wait
int potr_send_queue_push_wait(PotrSendQueue *q, PotrPeerId peer_id, uint16_t flags, const void *payload, uint16_t payload_len, volatile int *running)
Definition potrSendQueue.c:126
POTR_MUTEX_DESTROY
#define POTR_MUTEX_DESTROY(m)
Definition potrSendQueue.c:32
potr_send_queue_wait_drained
void potr_send_queue_wait_drained(PotrSendQueue *q)
Definition potrSendQueue.c:280
potrSendQueue.h
非同期送信キューの型定義と操作関数。
PotrPayloadElem
送信キューの 1 エントリ。ペイロードエレメント 1 個分のデータを保持する。
Definition potrSendQueue.h:51
PotrPayloadElem::payload_len
uint16_t payload_len
ペイロード長 (バイト)。
Definition potrSendQueue.h:54
PotrPayloadElem::flags
uint16_t flags
ペイロードエレメントフラグ (MORE_FRAG, COMPRESSED など)。
Definition potrSendQueue.h:53
PotrPayloadElem::payload
uint8_t * payload
ペイロードデータへのポインタ (プールスロット内を指す)。
Definition potrSendQueue.h:55
PotrPayloadElem::peer_id
PotrPeerId peer_id
送信先ピア識別子 (N:1 モード用。1:1 モードでは 0)。
Definition potrSendQueue.h:52
PotrSendQueue
非同期送信キュー。
Definition potrSendQueue.h:72
PotrSendQueue::depth
size_t depth
キュー容量 (エントリ数)。
Definition potrSendQueue.h:75
PotrSendQueue::count
size_t count
キュー内エントリ数。
Definition potrSendQueue.h:78
PotrSendQueue::inflight
size_t inflight
sendto 実行中エントリ数。
Definition potrSendQueue.h:79
PotrSendQueue::mutex
PotrMutex mutex
排他制御。
Definition potrSendQueue.h:80
PotrSendQueue::not_empty
PotrCondVar not_empty
count > 0 になったことを通知する条件変数。
Definition potrSendQueue.h:81
PotrSendQueue::drained
PotrCondVar drained
count == 0 && inflight == 0 を通知する条件変数。
Definition potrSendQueue.h:83
PotrSendQueue::head
size_t head
読み出し位置 (送信スレッドが使用)。
Definition potrSendQueue.h:76
PotrSendQueue::tail
size_t tail
書き込み位置 (potrSend 呼び出し元が使用)。
Definition potrSendQueue.h:77
PotrSendQueue::not_full
PotrCondVar not_full
count + inflight < depth になったことを通知する条件変数。
Definition potrSendQueue.h:82
PotrSendQueue::payload_pool
uint8_t * payload_pool
ペイロードプール (動的確保。depth × max_payload バイト)。
Definition potrSendQueue.h:74
PotrSendQueue::entries
PotrPayloadElem * entries
ペイロードエレメントバッファ (動的確保。depth 要素)。
Definition potrSendQueue.h:73
running
static volatile sig_atomic_t running
Definition tcpServer_linux.c:40