Document of c-modernization-kit (porter) 1.0.0

Loading...
Searching...
No Matches
potrOpenService.c
Go to the documentation of this file.
1
13
14#include <stdlib.h>
15#include <inttypes.h>
16#include <string.h>
17
18#ifndef _WIN32
19 #include <sys/socket.h>
20 #include <netinet/in.h>
21 #include <arpa/inet.h>
22 #include <unistd.h>
23 #include <time.h>
24#else /* _WIN32 */
25 #include <winsock2.h>
26 #include <ws2tcpip.h>
27#endif /* _WIN32 */
28
29#include <porter_const.h>
30#include <porter.h>
31
32#include "../protocol/packet.h"
33#include "../protocol/window.h"
34#include "../potrContext.h"
35#include "../potrPeerTable.h"
36#include "../thread/potrRecvThread.h"
37#include "../thread/potrHealthThread.h"
38#include "../thread/potrConnectThread.h"
39#include "../infra/compress/compress.h"
40#include "../infra/potrSendQueue.h"
41#include "../thread/potrSendThread.h"
42#include "../util/potrIpAddr.h"
43#include "../infra/potrLog.h"
44
45/* ソケットを作成して bind する。成功時は PotrSocket を返す。失敗時は POTR_INVALID_SOCKET。
46 bind_addr: bind する IPv4 アドレス。port: bind するポート番号 (0 = OS 自動選定)。 */
47 static PotrSocket open_socket_unicast(struct in_addr bind_addr, uint16_t port)
48{
49 PotrSocket sock;
50 struct sockaddr_in addr;
51 int reuse = 1;
52
53 sock = socket(AF_INET, SOCK_DGRAM, 0);
54 if (sock == POTR_INVALID_SOCKET)
55 {
56 return POTR_INVALID_SOCKET;
57 }
58
59#ifndef _WIN32
60 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
61#else /* _WIN32 */
62 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
63 (const char *)&reuse, sizeof(reuse));
64#endif /* _WIN32 */
65
66 memset(&addr, 0, sizeof(addr));
67 addr.sin_family = AF_INET;
68 addr.sin_addr = bind_addr;
69 addr.sin_port = htons(port);
70
71 if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0)
72 {
73#ifndef _WIN32
74 close(sock);
75#else /* _WIN32 */
76 closesocket(sock);
77#endif /* _WIN32 */
78 return POTR_INVALID_SOCKET;
79 }
80
81 return sock;
82}
83
84/* セッション識別子と開始時刻を生成してコンテキストに格納する。 */
85 static void generate_session(struct PotrContext_ *ctx)
86{
87#ifndef _WIN32
88 struct timespec ts;
89
90 srand((unsigned)((unsigned long)time(NULL) ^ (unsigned long)getpid()));
91 ctx->session_id = (uint32_t)rand();
92
93 clock_gettime(CLOCK_REALTIME, &ts);
94 ctx->session_tv_sec = (int64_t)ts.tv_sec;
95 ctx->session_tv_nsec = (int32_t)ts.tv_nsec;
96#else /* _WIN32 */
97 FILETIME ft;
98 ULARGE_INTEGER uli;
99
100 srand((unsigned)(GetTickCount() ^ GetCurrentProcessId()));
101 ctx->session_id = (uint32_t)rand();
102
103 GetSystemTimeAsFileTime(&ft);
104 uli.LowPart = ft.dwLowDateTime;
105 uli.HighPart = ft.dwHighDateTime;
106 /* FILETIME: 100ns 単位、1601-01-01 起点。Unix エポックとの差: 11644473600 秒 */
107 ctx->session_tv_sec = (int64_t)(uli.QuadPart / 10000000ULL) - 11644473600LL;
108 ctx->session_tv_nsec = (int32_t)((uli.QuadPart % 10000000ULL) * 100ULL);
109#endif /* _WIN32 */
110}
111
112/* マルチキャストソケットを作成して bind・グループ参加する。
113 src_if: 使用するローカルインターフェース (INADDR_ANY = OS 自動選択)。
114 is_receiver: 1 = 受信者、0 = 送信者。 */
115 static PotrSocket open_socket_multicast(const PotrServiceDef *def,
116 struct in_addr src_if,
117 int is_receiver)
118{
119 PotrSocket sock;
120 struct sockaddr_in addr;
121 struct ip_mreq mreq;
122 int reuse = 1;
123 /* 受信者: dst_port で bind する。送信者: src_port で bind する (送信元ポート)。 */
124 uint16_t bind_port;
125 if (is_receiver)
126 {
127 bind_port = def->dst_port;
128 }
129 else
130 {
131 bind_port = def->src_port;
132 }
133
134 sock = socket(AF_INET, SOCK_DGRAM, 0);
135 if (sock == POTR_INVALID_SOCKET)
136 {
137 return POTR_INVALID_SOCKET;
138 }
139
140#ifndef _WIN32
141 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
142#else /* _WIN32 */
143 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
144 (const char *)&reuse, sizeof(reuse));
145#endif /* _WIN32 */
146
147 memset(&addr, 0, sizeof(addr));
148 addr.sin_family = AF_INET;
149 addr.sin_addr.s_addr = htonl(INADDR_ANY);
150 addr.sin_port = htons(bind_port);
151
152 if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0)
153 {
154#ifndef _WIN32
155 close(sock);
156#else /* _WIN32 */
157 closesocket(sock);
158#endif /* _WIN32 */
159 return POTR_INVALID_SOCKET;
160 }
161
162 /* マルチキャストグループへ参加 (送受信ともに参加する) */
163 memset(&mreq, 0, sizeof(mreq));
164 if (parse_ipv4_addr(def->multicast_group, &mreq.imr_multiaddr) != POTR_SUCCESS)
165 {
166#ifndef _WIN32
167 close(sock);
168#else /* _WIN32 */
169 closesocket(sock);
170#endif /* _WIN32 */
171 return POTR_INVALID_SOCKET;
172 }
173 mreq.imr_interface = src_if;
174
175#ifndef _WIN32
176 if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,
177 &mreq, sizeof(mreq)) < 0)
178 {
179 close(sock);
180 return POTR_INVALID_SOCKET;
181 }
182#else /* _WIN32 */
183 if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,
184 (const char *)&mreq, sizeof(mreq)) < 0)
185 {
186 closesocket(sock);
187 return POTR_INVALID_SOCKET;
188 }
189#endif /* _WIN32 */
190
191 /* 送信者: マルチキャスト送信インターフェースを設定する */
192 if (!is_receiver)
193 {
194#ifndef _WIN32
195 setsockopt(sock, IPPROTO_IP, IP_MULTICAST_IF,
196 &src_if, sizeof(src_if));
197#else /* _WIN32 */
198 setsockopt(sock, IPPROTO_IP, IP_MULTICAST_IF,
199 (const char *)&src_if, sizeof(src_if));
200#endif /* _WIN32 */
201 }
202
203 return sock;
204}
205
206/* ブロードキャストソケットを作成して bind する。
207 src_port: 送信者の送信元 bind ポート (0 = OS 自動選定)。
208 dst_port: 受信者の listen ポート / 送信者の送信先ポート (省略不可)。
209 src_if: 送信者が使用するローカルインターフェース (INADDR_ANY = OS 自動選択)。
210 is_receiver: 1 = 受信者 (INADDR_ANY で bind)、0 = 送信者 (src_if で bind)。 */
211 static PotrSocket open_socket_broadcast(uint16_t src_port,
212 uint16_t dst_port,
213 struct in_addr src_if,
214 int is_receiver)
215{
216 PotrSocket sock;
217 struct sockaddr_in addr;
218 int reuse = 1;
219 int bcast = 1;
220 /* 受信者: dst_port で bind する。送信者: src_port で bind する (送信元ポート)。 */
221 uint16_t bind_port;
222 if (is_receiver)
223 {
224 bind_port = dst_port;
225 }
226 else
227 {
228 bind_port = src_port;
229 }
230
231 sock = socket(AF_INET, SOCK_DGRAM, 0);
232 if (sock == POTR_INVALID_SOCKET)
233 {
234 return POTR_INVALID_SOCKET;
235 }
236
237#ifndef _WIN32
238 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
239 setsockopt(sock, SOL_SOCKET, SO_BROADCAST, &bcast, sizeof(bcast));
240#else /* _WIN32 */
241 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
242 (const char *)&reuse, sizeof(reuse));
243 setsockopt(sock, SOL_SOCKET, SO_BROADCAST,
244 (const char *)&bcast, sizeof(bcast));
245#endif /* _WIN32 */
246
247 memset(&addr, 0, sizeof(addr));
248 addr.sin_family = AF_INET;
249 /* 送信者: src_addr で bind してインターフェースを選択する。受信者: INADDR_ANY で bind する。 */
250 if (!is_receiver)
251 {
252 addr.sin_addr.s_addr = src_if.s_addr;
253 }
254 else
255 {
256 addr.sin_addr.s_addr = htonl(INADDR_ANY);
257 }
258 addr.sin_port = htons(bind_port);
259
260 if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0)
261 {
262#ifndef _WIN32
263 close(sock);
264#else /* _WIN32 */
265 closesocket(sock);
266#endif /* _WIN32 */
267 return POTR_INVALID_SOCKET;
268 }
269
270 return sock;
271}
272
273/* 生成済みソケットをすべてクローズする */
274 static void cleanup_sockets(struct PotrContext_ *ctx)
275{
276 int i;
277 for (i = 0; i < (int)POTR_MAX_PATH; i++)
278 {
279 if (ctx->sock[i] != POTR_INVALID_SOCKET)
280 {
281#ifndef _WIN32
282 close(ctx->sock[i]);
283#else /* _WIN32 */
284 closesocket(ctx->sock[i]);
285#endif /* _WIN32 */
286 ctx->sock[i] = POTR_INVALID_SOCKET;
287 }
288 }
289}
290
291/* コンテキストが保持するすべてのリソースを解放して ctx 本体を free する。
292 memset(ctx, 0, ...) 後であれば、未初期化ポインタ (NULL) に対しても安全に呼び出せる。 */
293 static void ctx_cleanup(struct PotrContext_ *ctx)
294{
295 window_destroy(&ctx->send_window);
296 window_destroy(&ctx->recv_window);
297 free(ctx->frag_buf);
298 free(ctx->compress_buf);
299 free(ctx->crypto_buf);
300 free(ctx->recv_buf);
301 free(ctx->send_wire_buf);
302 if (ctx->is_multi_peer && ctx->peers != NULL)
303 {
304 peer_table_destroy(ctx);
305 }
306 /* TCP listen ソケットをクローズ (path ごと) */
307 {
308 int i;
309 for (i = 0; i < (int)POTR_MAX_PATH; i++)
310 {
311 if (ctx->tcp_listen_sock[i] != POTR_INVALID_SOCKET)
312 {
313#ifndef _WIN32
314 close(ctx->tcp_listen_sock[i]);
315#else /* _WIN32 */
316 closesocket(ctx->tcp_listen_sock[i]);
317#endif /* _WIN32 */
318 }
319 }
320 }
321 {
322 int i;
323 for (i = 0; i < (int)POTR_MAX_PATH; i++)
324 {
325 if (ctx->tcp_conn_fd[i] != POTR_INVALID_SOCKET)
326 {
327#ifndef _WIN32
328 close(ctx->tcp_conn_fd[i]);
329#else /* _WIN32 */
330 closesocket(ctx->tcp_conn_fd[i]);
331#endif /* _WIN32 */
332 }
333 }
334 }
335 cleanup_sockets(ctx);
336 free(ctx);
337}
338
339/* TCP RECEIVER: path_idx 番目の listen ソケットを作成して bind・listen する。
340 dst_addr[path_idx] が指定されていれば dst_addr_resolved[path_idx] に解決する。
341 src_addr[path_idx] が指定されていれば src_addr_resolved[path_idx] にも解決する（接続元フィルタ用）。
342 成功時は ctx->tcp_listen_sock[path_idx] に格納して POTR_SUCCESS を返す。 */
343 static int open_socket_tcp_receiver(struct PotrContext_ *ctx, int path_idx)
344{
345 PotrSocket sock;
346 struct sockaddr_in addr;
347 int reuse = 1;
348 struct in_addr bind_ip;
349
350 if (ctx->service.dst_addr[path_idx][0] != '\0')
351 {
352 if (resolve_ipv4_addr(ctx->service.dst_addr[path_idx], &bind_ip) != POTR_SUCCESS)
353 {
354 return POTR_ERROR;
355 }
356 ctx->dst_addr_resolved[path_idx] = bind_ip;
357 }
358 else
359 {
360 bind_ip.s_addr = htonl(INADDR_ANY);
361 }
362
363 if (ctx->service.src_addr[path_idx][0] != '\0')
364 {
365 if (resolve_ipv4_addr(ctx->service.src_addr[path_idx],
366 &ctx->src_addr_resolved[path_idx]) != POTR_SUCCESS)
367 {
368 return POTR_ERROR;
369 }
370 }
371
372 sock = socket(AF_INET, SOCK_STREAM, 0);
373 if (sock == POTR_INVALID_SOCKET)
374 {
375 return POTR_ERROR;
376 }
377
378#ifndef _WIN32
379 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
380#else /* _WIN32 */
381 setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
382 (const char *)&reuse, sizeof(reuse));
383#endif /* _WIN32 */
384
385 memset(&addr, 0, sizeof(addr));
386 addr.sin_family = AF_INET;
387 addr.sin_addr = bind_ip;
388 addr.sin_port = htons(ctx->service.dst_port);
389
390 if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0)
391 {
392#ifndef _WIN32
393 close(sock);
394#else /* _WIN32 */
395 closesocket(sock);
396#endif /* _WIN32 */
397 return POTR_ERROR;
398 }
399
400 if (listen(sock, SOMAXCONN) < 0)
401 {
402#ifndef _WIN32
403 close(sock);
404#else /* _WIN32 */
405 closesocket(sock);
406#endif /* _WIN32 */
407 return POTR_ERROR;
408 }
409
410 ctx->tcp_listen_sock[path_idx] = sock;
411 return POTR_SUCCESS;
412}
413
414/* TCP SENDER: path_idx 番目の接続先 dst_addr を解決して dst_addr_resolved[path_idx] に格納する。
415 src_addr[path_idx] が指定されていれば src_addr_resolved[path_idx] にも解決する。
416 実際の TCP 接続は connect スレッドが行う。 */
417 static int open_socket_tcp_sender(struct PotrContext_ *ctx, int path_idx)
418{
419 if (ctx->service.dst_addr[path_idx][0] == '\0')
420 {
421 POTR_LOG(POTR_TRACE_ERROR,
422 "open_socket_tcp_sender: dst_addr[%d] is empty", path_idx);
423 return POTR_ERROR;
424 }
425
426 if (resolve_ipv4_addr(ctx->service.dst_addr[path_idx],
427 &ctx->dst_addr_resolved[path_idx]) != POTR_SUCCESS)
428 {
429 return POTR_ERROR;
430 }
431
432 if (ctx->service.src_addr[path_idx][0] != '\0')
433 {
434 if (resolve_ipv4_addr(ctx->service.src_addr[path_idx],
435 &ctx->src_addr_resolved[path_idx]) != POTR_SUCCESS)
436 {
437 return POTR_ERROR;
438 }
439 }
440
441 return POTR_SUCCESS;
442}
443
444/* doxygen コメントは、ヘッダに記載 */
445 POTR_EXPORT int POTR_API potrOpenService(const PotrGlobalConfig *global,
446 const PotrServiceDef *service,
447 PotrRole role,
448 PotrRecvCallback callback,
449 PotrHandle *handle)
450{
451 struct PotrContext_ *ctx;
452
453 if (global == NULL || service == NULL || handle == NULL)
454 {
455 POTR_LOG(POTR_TRACE_ERROR,
456 "potrOpenService: invalid argument (global=%p service=%p handle=%p)",
457 (const void *)global, (const void *)service, (const void *)handle);
458 return POTR_ERROR;
459 }
460
461 POTR_LOG(POTR_TRACE_VERBOSE,
462 "potrOpenService: service_id=%" PRId64 " role=%d",
463 service->service_id, (int)role);
464
465 /* role と callback の整合性チェック (設定読み込み前に確定できる部分のみ) */
466 if (role == POTR_ROLE_RECEIVER && callback == NULL)
467 {
468 POTR_LOG(POTR_TRACE_ERROR,
469 "potrOpenService: service_id=%" PRId64 " RECEIVER role requires callback",
470 service->service_id);
471 return POTR_ERROR;
472 }
473 /* SENDER + callback の完全チェックは設定読み込み後に行う
474 (POTR_TYPE_UNICAST_BIDIR の SENDER は callback が必須のため) */
475 if (role != POTR_ROLE_SENDER && role != POTR_ROLE_RECEIVER)
476 {
477 POTR_LOG(POTR_TRACE_ERROR,
478 "potrOpenService: service_id=%" PRId64 " unknown role=%d",
479 service->service_id, (int)role);
480 return POTR_ERROR;
481 }
482
483#ifdef _WIN32
484 {
485 WSADATA wsa;
486 if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0)
487 {
488 return POTR_ERROR;
489 }
490 }
491#endif /* _WIN32 */
492
493 ctx = (struct PotrContext_ *)malloc(sizeof(struct PotrContext_));
494 if (ctx == NULL)
495 {
496 return POTR_ERROR;
497 }
498 memset(ctx, 0, sizeof(*ctx));
499
500 /* 全ソケットを INVALID で初期化 */
501 {
502 int i;
503 for (i = 0; i < (int)POTR_MAX_PATH; i++)
504 {
505 ctx->sock[i] = POTR_INVALID_SOCKET;
506 ctx->tcp_conn_fd[i] = POTR_INVALID_SOCKET;
507 ctx->tcp_listen_sock[i] = POTR_INVALID_SOCKET;
508 }
509 }
510
511 /* グローバル設定とサービス定義をコンテキストにコピー */
512 memcpy(&ctx->global, global, sizeof(PotrGlobalConfig));
513 memcpy(&ctx->service, service, sizeof(PotrServiceDef));
514
515 /* SENDER + callback の整合性チェック (型が確定した後) */
516 if (role == POTR_ROLE_SENDER && callback != NULL
517 && ctx->service.type != POTR_TYPE_UNICAST_BIDIR)
518 {
519 POTR_LOG(POTR_TRACE_ERROR,
520 "potrOpenService: service_id=%" PRId64 " SENDER role must not have callback"
521 " (type=%d)",
522 ctx->service.service_id, (int)ctx->service.type);
523 free(ctx);
524 return POTR_ERROR;
525 }
526 if (role == POTR_ROLE_SENDER && callback == NULL
527 && ctx->service.type == POTR_TYPE_UNICAST_BIDIR)
528 {
529 POTR_LOG(POTR_TRACE_ERROR,
530 "potrOpenService: service_id=%" PRId64 " UNICAST_BIDIR SENDER role requires callback",
531 ctx->service.service_id);
532 free(ctx);
533 return POTR_ERROR;
534 }
535
536 /* 設定値バリデーション */
537 if (ctx->global.max_payload < 64U || ctx->global.max_payload > POTR_MAX_PAYLOAD)
538 {
539 POTR_LOG(POTR_TRACE_ERROR,
540 "potrOpenService: service_id=%" PRId64 " invalid max_payload=%u (range: 64..%u)",
541 ctx->service.service_id, (unsigned)ctx->global.max_payload, (unsigned)POTR_MAX_PAYLOAD);
542 free(ctx);
543 return POTR_ERROR;
544 }
545 if (ctx->global.window_size < 2U || ctx->global.window_size > POTR_MAX_WINDOW_SIZE)
546 {
547 POTR_LOG(POTR_TRACE_ERROR,
548 "potrOpenService: service_id=%" PRId64 " invalid window_size=%u (range: 2..%u)",
549 ctx->service.service_id, (unsigned)ctx->global.window_size, (unsigned)POTR_MAX_WINDOW_SIZE);
550 free(ctx);
551 return POTR_ERROR;
552 }
553 if (ctx->global.max_message_size < (uint32_t)ctx->global.max_payload)
554 {
555 POTR_LOG(POTR_TRACE_ERROR,
556 "potrOpenService: service_id=%" PRId64 " max_message_size=%u must be >= max_payload=%u",
557 ctx->service.service_id, (unsigned)ctx->global.max_message_size,
558 (unsigned)ctx->global.max_payload);
559 free(ctx);
560 return POTR_ERROR;
561 }
562 if (ctx->global.send_queue_depth < 2U)
563 {
564 POTR_LOG(POTR_TRACE_ERROR,
565 "potrOpenService: service_id=%" PRId64 " invalid send_queue_depth=%u (min: 2)",
566 ctx->service.service_id, (unsigned)ctx->global.send_queue_depth);
567 free(ctx);
568 return POTR_ERROR;
569 }
570
571 /* TCP 型: グローバルヘルス設定を TCP 専用値で上書きする。
572 サービスレベル設定 (0 以外) はさらにそれを上書きする。 */
573 if (potr_is_tcp_type(ctx->service.type))
574 {
575 ctx->global.health_interval_ms = ctx->global.tcp_health_interval_ms;
576 ctx->global.health_timeout_ms = ctx->global.tcp_health_timeout_ms;
577 }
578 if (ctx->service.health_interval_ms != 0U)
579 {
580 ctx->global.health_interval_ms = ctx->service.health_interval_ms;
581 }
582 if (ctx->service.health_timeout_ms != 0U)
583 {
584 ctx->global.health_timeout_ms = ctx->service.health_timeout_ms;
585 }
586
587 POTR_LOG(POTR_TRACE_VERBOSE,
588 "potrOpenService: service_id=%" PRId64 " type=%d window=%u max_payload=%u"
589 " max_message_size=%u send_queue_depth=%u"
590 " health_interval=%ums health_timeout=%ums",
591 ctx->service.service_id, (int)ctx->service.type,
592 (unsigned)ctx->global.window_size, (unsigned)ctx->global.max_payload,
593 (unsigned)ctx->global.max_message_size, (unsigned)ctx->global.send_queue_depth,
594 (unsigned)ctx->global.health_interval_ms,
595 (unsigned)ctx->global.health_timeout_ms);
596
597 /* 通信種別に応じてソケットを作成 (RAW 型はベース型に正規化してから判定) */
598 switch (potr_raw_base_type(ctx->service.type))
599 {
600 case POTR_TYPE_UNICAST:
601 {
602 int i;
603
604 if (ctx->service.dst_port == 0)
605 {
606 free(ctx);
607 return POTR_ERROR;
608 }
609
610 for (i = 0; i < (int)POTR_MAX_PATH; i++)
611 {
612 struct in_addr bind_addr;
613 uint16_t bind_port;
614
615 if (ctx->service.src_addr[i][0] == '\0' ||
616 ctx->service.dst_addr[i][0] == '\0')
617 {
618 break;
619 }
620
621 if (resolve_ipv4_addr(ctx->service.src_addr[i],
622 &ctx->src_addr_resolved[i]) != POTR_SUCCESS ||
623 resolve_ipv4_addr(ctx->service.dst_addr[i],
624 &ctx->dst_addr_resolved[i]) != POTR_SUCCESS)
625 {
626 ctx_cleanup(ctx);
627 return POTR_ERROR;
628 }
629
630 if (role == POTR_ROLE_RECEIVER)
631 {
632 bind_addr = ctx->dst_addr_resolved[i];
633 bind_port = ctx->service.dst_port;
634 }
635 else
636 {
637 bind_addr = ctx->src_addr_resolved[i];
638 bind_port = ctx->service.src_port;
639 }
640
641 ctx->sock[i] = open_socket_unicast(bind_addr, bind_port);
642 if (ctx->sock[i] == POTR_INVALID_SOCKET)
643 {
644 ctx_cleanup(ctx);
645 return POTR_ERROR;
646 }
647
648 ctx->n_path++;
649 }
650
651 if (ctx->n_path == 0)
652 {
653 ctx_cleanup(ctx);
654 return POTR_ERROR;
655 }
656 break;
657 }
658
659 case POTR_TYPE_MULTICAST:
660 {
661 int i;
662
663 if (ctx->service.dst_port == 0 ||
664 ctx->service.multicast_group[0] == '\0')
665 {
666 ctx_cleanup(ctx);
667 return POTR_ERROR;
668 }
669
670 for (i = 0; i < (int)POTR_MAX_PATH; i++)
671 {
672 if (ctx->service.src_addr[i][0] == '\0') break;
673
674 if (resolve_ipv4_addr(ctx->service.src_addr[i],
675 &ctx->src_addr_resolved[i]) != POTR_SUCCESS)
676 {
677 ctx_cleanup(ctx);
678 return POTR_ERROR;
679 }
680
681 ctx->sock[i] = open_socket_multicast(&ctx->service,
682 ctx->src_addr_resolved[i],
683 role == POTR_ROLE_RECEIVER);
684 if (ctx->sock[i] == POTR_INVALID_SOCKET)
685 {
686 ctx_cleanup(ctx);
687 return POTR_ERROR;
688 }
689
690 ctx->n_path++;
691 }
692
693 if (ctx->n_path == 0)
694 {
695 ctx_cleanup(ctx);
696 return POTR_ERROR;
697 }
698 break;
699 }
700
701 case POTR_TYPE_BROADCAST:
702 {
703 int i;
704
705 if (ctx->service.dst_port == 0)
706 {
707 ctx_cleanup(ctx);
708 return POTR_ERROR;
709 }
710
711 /* broadcast_addr 省略時は限定ブロードキャスト (255.255.255.255) を使用する */
712 if (ctx->service.broadcast_addr[0] == '\0')
713 {
714 const char *dflt = "255.255.255.255";
715 size_t len = strlen(dflt);
716 memcpy(ctx->service.broadcast_addr, dflt, len + 1);
717 }
718
719 for (i = 0; i < (int)POTR_MAX_PATH; i++)
720 {
721 if (ctx->service.src_addr[i][0] == '\0') break;
722
723 if (resolve_ipv4_addr(ctx->service.src_addr[i],
724 &ctx->src_addr_resolved[i]) != POTR_SUCCESS)
725 {
726 ctx_cleanup(ctx);
727 return POTR_ERROR;
728 }
729
730 ctx->sock[i] = open_socket_broadcast(ctx->service.src_port,
731 ctx->service.dst_port,
732 ctx->src_addr_resolved[i],
733 role == POTR_ROLE_RECEIVER);
734 if (ctx->sock[i] == POTR_INVALID_SOCKET)
735 {
736 ctx_cleanup(ctx);
737 return POTR_ERROR;
738 }
739
740 ctx->n_path++;
741 }
742
743 if (ctx->n_path == 0)
744 {
745 ctx_cleanup(ctx);
746 return POTR_ERROR;
747 }
748 break;
749 }
750
751 case POTR_TYPE_UNICAST_BIDIR:
752 {
753 ctx->is_multi_peer = 0;
754
755 /* dst_port は必須。 */
756 if (ctx->service.dst_port == 0)
757 {
758 POTR_LOG(POTR_TRACE_ERROR,
759 "potrOpenService: service_id=%" PRId64 " UNICAST_BIDIR requires"
760 " dst_port (non-zero)",
761 ctx->service.service_id);
762 ctx_cleanup(ctx);
763 return POTR_ERROR;
764 }
765
766 if (role == POTR_ROLE_RECEIVER && ctx->service.src_addr[0][0] == '\0')
767 {
768 /* 動的 1:1 RECEIVER: src_addr 省略 → dst_addr:dst_port に bind し、
769 最初の受信パケットから SENDER のアドレスを動的学習する。 */
770 struct in_addr bind_addr;
771
772 if (ctx->service.dst_addr[0][0] == '\0')
773 {
774 bind_addr.s_addr = htonl(INADDR_ANY);
775 }
776 else
777 {
778 if (resolve_ipv4_addr(ctx->service.dst_addr[0],
779 &bind_addr) != POTR_SUCCESS)
780 {
781 ctx_cleanup(ctx);
782 return POTR_ERROR;
783 }
784 ctx->dst_addr_resolved[0] = bind_addr;
785 }
786 ctx->sock[0] = open_socket_unicast(bind_addr, ctx->service.dst_port);
787 if (ctx->sock[0] == POTR_INVALID_SOCKET)
788 {
789 ctx_cleanup(ctx);
790 return POTR_ERROR;
791 }
792 ctx->n_path = 1;
793 POTR_LOG(POTR_TRACE_INFO,
794 "potrOpenService: service_id=%" PRId64 " UNICAST_BIDIR 1:1 dynamic RECEIVER"
795 " bind dst_port=%u",
796 ctx->service.service_id, (unsigned)ctx->service.dst_port);
797 }
798 else
799 {
800 /* 1:1 モード: src_addr/dst_addr ペアループ。
801 SENDER は src_addr 省略時に INADDR_ANY で bind する (OS がアダプタを自動選択)。
802 RECEIVER はここには src_addr がある場合のみ到達する。 */
803 int i;
804
805 for (i = 0; i < (int)POTR_MAX_PATH; i++)
806 {
807 struct in_addr bind_addr;
808
809 /* dst_addr が空 → パス終端。
810 RECEIVER は src_addr も必要 (src_addr なし RECEIVER は上で処理済み)。 */
811 if (ctx->service.dst_addr[i][0] == '\0') break;
812 if (role == POTR_ROLE_RECEIVER
813 && ctx->service.src_addr[i][0] == '\0') break;
814
815 if (ctx->service.src_addr[i][0] != '\0')
816 {
817 if (resolve_ipv4_addr(ctx->service.src_addr[i],
818 &ctx->src_addr_resolved[i]) != POTR_SUCCESS)
819 {
820 ctx_cleanup(ctx);
821 return POTR_ERROR;
822 }
823 }
824
825 if (resolve_ipv4_addr(ctx->service.dst_addr[i],
826 &ctx->dst_addr_resolved[i]) != POTR_SUCCESS)
827 {
828 ctx_cleanup(ctx);
829 return POTR_ERROR;
830 }
831
832 if (role == POTR_ROLE_SENDER)
833 {
834 /* src_addr 省略時は INADDR_ANY で bind し OS がアダプタを自動選択 */
835 if (ctx->service.src_addr[i][0] != '\0')
836 bind_addr = ctx->src_addr_resolved[i];
837 else
838 bind_addr.s_addr = htonl(INADDR_ANY);
839 ctx->sock[i] = open_socket_unicast(bind_addr, ctx->service.src_port);
840 }
841 else
842 {
843 /* RECEIVER: dst_addr:dst_port で bind */
844 ctx->sock[i] = open_socket_unicast(ctx->dst_addr_resolved[i],
845 ctx->service.dst_port);
846 }
847 if (ctx->sock[i] == POTR_INVALID_SOCKET)
848 {
849 ctx_cleanup(ctx);
850 return POTR_ERROR;
851 }
852
853 ctx->n_path++;
854 }
855
856 if (ctx->n_path == 0)
857 {
858 ctx_cleanup(ctx);
859 return POTR_ERROR;
860 }
861 }
862 break;
863 }
864
865 case POTR_TYPE_UNICAST_BIDIR_N1:
866 {
867 /* N:1 サーバ: dst_addr[i]:dst_port ごとにソケットを bind する。
868 dst_addr が全て省略されている場合は INADDR_ANY で 1 ソケットのみ作成する。 */
869 int i;
870
871 ctx->is_multi_peer = 1;
872
873 /* dst_port は必須。 */
874 if (ctx->service.dst_port == 0)
875 {
876 POTR_LOG(POTR_TRACE_ERROR,
877 "potrOpenService: service_id=%" PRId64 " UNICAST_BIDIR_N1 requires"
878 " dst_port (non-zero)",
879 ctx->service.service_id);
880 ctx_cleanup(ctx);
881 return POTR_ERROR;
882 }
883
884 if (ctx->service.dst_addr[0][0] == '\0')
885 {
886 /* dst_addr 全て省略: INADDR_ANY で 1 ソケット */
887 struct in_addr any_addr;
888 any_addr.s_addr = htonl(INADDR_ANY);
889 ctx->sock[0] = open_socket_unicast(any_addr, ctx->service.dst_port);
890 if (ctx->sock[0] == POTR_INVALID_SOCKET)
891 {
892 ctx_cleanup(ctx);
893 return POTR_ERROR;
894 }
895 ctx->n_path = 1;
896 }
897 else
898 {
899 /* dst_addr[i] を列挙してパスごとにソケットを作成する */
900 for (i = 0; i < (int)POTR_MAX_PATH; i++)
901 {
902 struct in_addr bind_addr;
903
904 if (ctx->service.dst_addr[i][0] == '\0') break;
905
906 if (resolve_ipv4_addr(ctx->service.dst_addr[i],
907 &bind_addr) != POTR_SUCCESS)
908 {
909 ctx_cleanup(ctx);
910 return POTR_ERROR;
911 }
912 ctx->dst_addr_resolved[i] = bind_addr;
913 ctx->sock[i] = open_socket_unicast(bind_addr, ctx->service.dst_port);
914 if (ctx->sock[i] == POTR_INVALID_SOCKET)
915 {
916 ctx_cleanup(ctx);
917 return POTR_ERROR;
918 }
919 ctx->n_path = i + 1;
920 }
921 }
922
923 /* ピアテーブル初期化 */
924 ctx->max_peers = (int)ctx->service.max_peers;
925 if (ctx->max_peers <= 0)
926 {
927 ctx->max_peers = 1024;
928 }
929 if (peer_table_init(ctx) != POTR_SUCCESS)
930 {
931 ctx_cleanup(ctx);
932 return POTR_ERROR;
933 }
934
935 POTR_LOG(POTR_TRACE_INFO,
936 "potrOpenService: service_id=%" PRId64 " UNICAST_BIDIR_N1"
937 " (max_peers=%d src_port_filter=%u) bind dst_port=%u n_path=%d",
938 ctx->service.service_id, ctx->max_peers,
939 (unsigned)ctx->service.src_port,
940 (unsigned)ctx->service.dst_port,
941 ctx->n_path);
942 break;
943 }
944
945 case POTR_TYPE_TCP:
946 case POTR_TYPE_TCP_BIDIR:
947 {
948 if (ctx->service.dst_port == 0)
949 {
950 POTR_LOG(POTR_TRACE_ERROR,
951 "potrOpenService: service_id=%" PRId64 " TCP requires dst_port",
952 ctx->service.service_id);
953 ctx_cleanup(ctx);
954 return POTR_ERROR;
955 }
956
957 if (role == POTR_ROLE_RECEIVER)
958 {
959 /* 非空の dst_addr[i] ごとに listen ソケットを作成する */
960 int i;
961 for (i = 0; i < (int)POTR_MAX_PATH; i++)
962 {
963 if (ctx->service.dst_addr[i][0] == '\0') break;
964
965 if (open_socket_tcp_receiver(ctx, i) != POTR_SUCCESS)
966 {
967 POTR_LOG(POTR_TRACE_ERROR,
968 "potrOpenService: service_id=%" PRId64 " TCP listen failed"
969 " (path=%d dst_addr=%s dst_port=%u)",
970 ctx->service.service_id, i,
971 ctx->service.dst_addr[i][0] ? ctx->service.dst_addr[i] : "*",
972 (unsigned)ctx->service.dst_port);
973 ctx_cleanup(ctx);
974 return POTR_ERROR;
975 }
976 POTR_LOG(POTR_TRACE_INFO,
977 "potrOpenService: service_id=%" PRId64 " TCP path[%d] listening"
978 " on %s:%u",
979 ctx->service.service_id, i,
980 ctx->service.dst_addr[i][0] ? ctx->service.dst_addr[i] : "*",
981 (unsigned)ctx->service.dst_port);
982 ctx->n_path = i + 1;
983 }
984 if (ctx->n_path == 0)
985 {
986 POTR_LOG(POTR_TRACE_ERROR,
987 "potrOpenService: service_id=%" PRId64 " TCP RECEIVER requires"
988 " at least one dst_addr",
989 ctx->service.service_id);
990 ctx_cleanup(ctx);
991 return POTR_ERROR;
992 }
993 }
994 else
995 {
996 /* 非空の dst_addr[i] ごとにアドレスを解決する */
997 int i;
998 for (i = 0; i < (int)POTR_MAX_PATH; i++)
999 {
1000 if (ctx->service.dst_addr[i][0] == '\0') break;
1001
1002 if (open_socket_tcp_sender(ctx, i) != POTR_SUCCESS)
1003 {
1004 POTR_LOG(POTR_TRACE_ERROR,
1005 "potrOpenService: service_id=%" PRId64 " TCP sender"
1006 " dst_addr resolve failed (path=%d %s)",
1007 ctx->service.service_id, i, ctx->service.dst_addr[i]);
1008 ctx_cleanup(ctx);
1009 return POTR_ERROR;
1010 }
1011 ctx->n_path = i + 1;
1012 }
1013 if (ctx->n_path == 0)
1014 {
1015 POTR_LOG(POTR_TRACE_ERROR,
1016 "potrOpenService: service_id=%" PRId64 " TCP SENDER requires"
1017 " at least one dst_addr",
1018 ctx->service.service_id);
1019 ctx_cleanup(ctx);
1020 return POTR_ERROR;
1021 }
1022 }
1023 break;
1024 }
1025
1026 case POTR_TYPE_UNICAST_RAW:
1027 case POTR_TYPE_MULTICAST_RAW:
1028 case POTR_TYPE_BROADCAST_RAW:
1029 /* potr_raw_base_type() は RAW 型をベース型に変換するため、ここには到達しない */
1030 /* fall through */
1031 default:
1032 ctx_cleanup(ctx);
1033 return POTR_ERROR;
1034 }
1035
1036 ctx->callback = callback;
1037 ctx->role = role;
1038
1039 /* 送信先ソケットアドレスを設定 (RAW 型はベース型に正規化してから判定)
1040 UNICAST_BIDIR は両端 (SENDER / RECEIVER) ともに dest_addr を設定する */
1041 if (role == POTR_ROLE_SENDER
1042 || ctx->service.type == POTR_TYPE_UNICAST_BIDIR)
1043 {
1044 int i;
1045
1046 switch (potr_raw_base_type(ctx->service.type))
1047 {
1048 case POTR_TYPE_UNICAST_BIDIR:
1049 case POTR_TYPE_UNICAST_BIDIR_N1:
1050 for (i = 0; i < ctx->n_path; i++)
1051 {
1052 memset(&ctx->dest_addr[i], 0, sizeof(ctx->dest_addr[i]));
1053 ctx->dest_addr[i].sin_family = AF_INET;
1054 if (role == POTR_ROLE_SENDER)
1055 {
1056 /* SENDER: dst_addr:dst_port (RECEIVER の bind アドレス) へ送信 */
1057 ctx->dest_addr[i].sin_addr = ctx->dst_addr_resolved[i];
1058 ctx->dest_addr[i].sin_port = htons(ctx->service.dst_port);
1059 }
1060 else
1061 {
1062 /* RECEIVER: src_addr:src_port (SENDER の bind アドレス) へ送信 */
1063 ctx->dest_addr[i].sin_addr = ctx->src_addr_resolved[i];
1064 ctx->dest_addr[i].sin_port = htons(ctx->service.src_port);
1065 }
1066 }
1067 break;
1068
1069 case POTR_TYPE_UNICAST:
1070 for (i = 0; i < ctx->n_path; i++)
1071 {
1072 memset(&ctx->dest_addr[i], 0, sizeof(ctx->dest_addr[i]));
1073 ctx->dest_addr[i].sin_family = AF_INET;
1074 ctx->dest_addr[i].sin_addr = ctx->dst_addr_resolved[i];
1075 ctx->dest_addr[i].sin_port = htons(ctx->service.dst_port);
1076 }
1077 break;
1078
1079 case POTR_TYPE_MULTICAST:
1080 {
1081 struct in_addr mcast_ip;
1082 if (parse_ipv4_addr(ctx->service.multicast_group, &mcast_ip) != POTR_SUCCESS)
1083 {
1084 ctx_cleanup(ctx);
1085 return POTR_ERROR;
1086 }
1087 for (i = 0; i < ctx->n_path; i++)
1088 {
1089 memset(&ctx->dest_addr[i], 0, sizeof(ctx->dest_addr[i]));
1090 ctx->dest_addr[i].sin_family = AF_INET;
1091 ctx->dest_addr[i].sin_addr = mcast_ip;
1092 ctx->dest_addr[i].sin_port = htons(ctx->service.dst_port);
1093 }
1094 break;
1095 }
1096
1097 case POTR_TYPE_BROADCAST:
1098 {
1099 struct in_addr bcast_ip;
1100 if (parse_ipv4_addr(ctx->service.broadcast_addr, &bcast_ip) != POTR_SUCCESS)
1101 {
1102 ctx_cleanup(ctx);
1103 return POTR_ERROR;
1104 }
1105 for (i = 0; i < ctx->n_path; i++)
1106 {
1107 memset(&ctx->dest_addr[i], 0, sizeof(ctx->dest_addr[i]));
1108 ctx->dest_addr[i].sin_family = AF_INET;
1109 ctx->dest_addr[i].sin_addr = bcast_ip;
1110 ctx->dest_addr[i].sin_port = htons(ctx->service.dst_port);
1111 }
1112 break;
1113 }
1114
1115 case POTR_TYPE_TCP:
1116 case POTR_TYPE_TCP_BIDIR:
1117 /* TCP 接続ソケットは connect スレッドが管理するため dest_addr 設定不要 */
1118 break;
1119 case POTR_TYPE_UNICAST_RAW:
1120 case POTR_TYPE_MULTICAST_RAW:
1121 case POTR_TYPE_BROADCAST_RAW:
1122 /* potr_raw_base_type() は RAW 型をベース型に変換するため、ここには到達しない */
1123 /* fall through */
1124 default:
1125 break;
1126 }
1127 }
1128
1129 /* セッション識別子を生成する */
1130 generate_session(ctx);
1131
1132 /* 送受信ウィンドウを初期化 */
1133 if (window_init(&ctx->send_window, 0,
1134 ctx->global.window_size, ctx->global.max_payload) != POTR_SUCCESS)
1135 {
1136 ctx_cleanup(ctx);
1137 return POTR_ERROR;
1138 }
1139 if (window_init(&ctx->recv_window, 0,
1140 ctx->global.window_size, ctx->global.max_payload) != POTR_SUCCESS)
1141 {
1142 ctx_cleanup(ctx);
1143 return POTR_ERROR;
1144 }
1145
1146 /* 動的バッファを確保 */
1147 ctx->frag_buf = (uint8_t *)malloc(ctx->global.max_message_size);
1148 if (ctx->frag_buf == NULL)
1149 {
1150 ctx_cleanup(ctx);
1151 return POTR_ERROR;
1152 }
1153
1154 ctx->compress_buf_size = POTR_COMPRESS_HEADER_SIZE
1155 + (size_t)ctx->global.max_message_size + 64U;
1156 ctx->compress_buf = (uint8_t *)malloc(ctx->compress_buf_size);
1157 if (ctx->compress_buf == NULL)
1158 {
1159 ctx_cleanup(ctx);
1160 return POTR_ERROR;
1161 }
1162
1163 ctx->recv_buf = (uint8_t *)malloc(PACKET_HEADER_SIZE + ctx->global.max_payload);
1164 if (ctx->recv_buf == NULL)
1165 {
1166 ctx_cleanup(ctx);
1167 return POTR_ERROR;
1168 }
1169
1170 ctx->send_wire_buf = (uint8_t *)malloc(PACKET_HEADER_SIZE + ctx->global.max_payload);
1171 if (ctx->send_wire_buf == NULL)
1172 {
1173 ctx_cleanup(ctx);
1174 return POTR_ERROR;
1175 }
1176
1177 ctx->crypto_buf_size = ctx->global.max_payload + POTR_CRYPTO_TAG_SIZE;
1178 ctx->crypto_buf = (uint8_t *)malloc(ctx->crypto_buf_size);
1179 if (ctx->crypto_buf == NULL)
1180 {
1181 ctx_cleanup(ctx);
1182 return POTR_ERROR;
1183 }
1184
1185 /* TCP: 接続管理スレッドを起動する。
1186 send/recv/health スレッドは接続確立後に connect スレッドが管理する。 */
1187 if (potr_is_tcp_type(ctx->service.type))
1188 {
1189 /* tcp_state_mutex / tcp_state_cv / tcp_send_mutex[] / recv_window_mutex /
1190 health_mutex[] / health_wakeup[] を初期化 */
1191#ifndef _WIN32
1192 {
1193 int i;
1194 pthread_mutex_init(&ctx->tcp_state_mutex, NULL);
1195 pthread_cond_init(&ctx->tcp_state_cv, NULL);
1196 for (i = 0; i < (int)POTR_MAX_PATH; i++)
1197 {
1198 pthread_mutex_init(&ctx->tcp_send_mutex[i], NULL);
1199 pthread_mutex_init(&ctx->health_mutex[i], NULL);
1200 pthread_cond_init(&ctx->health_wakeup[i], NULL);
1201 }
1202 pthread_mutex_init(&ctx->recv_window_mutex, NULL);
1203 }
1204#else /* _WIN32 */
1205 {
1206 int i;
1207 InitializeCriticalSection(&ctx->tcp_state_mutex);
1208 InitializeConditionVariable(&ctx->tcp_state_cv);
1209 for (i = 0; i < (int)POTR_MAX_PATH; i++)
1210 {
1211 InitializeCriticalSection(&ctx->tcp_send_mutex[i]);
1212 InitializeCriticalSection(&ctx->health_mutex[i]);
1213 InitializeConditionVariable(&ctx->health_wakeup[i]);
1214 }
1215 InitializeCriticalSection(&ctx->recv_window_mutex);
1216 }
1217#endif /* _WIN32 */
1218
1219 /* SENDER または TCP_BIDIR: 送信キューを初期化 (connect スレッドが reconnect 時に destroy+init する) */
1220 if (role == POTR_ROLE_SENDER
1221 || ctx->service.type == POTR_TYPE_TCP_BIDIR)
1222 {
1223 if (potr_send_queue_init(&ctx->send_queue,
1224 (size_t)ctx->global.send_queue_depth,
1225 ctx->global.max_payload) != POTR_SUCCESS)
1226 {
1227#ifndef _WIN32
1228 {
1229 int i;
1230 pthread_mutex_destroy(&ctx->tcp_state_mutex);
1231 pthread_cond_destroy(&ctx->tcp_state_cv);
1232 for (i = 0; i < (int)POTR_MAX_PATH; i++)
1233 pthread_mutex_destroy(&ctx->tcp_send_mutex[i]);
1234 pthread_mutex_destroy(&ctx->recv_window_mutex);
1235 }
1236#else /* _WIN32 */
1237 {
1238 int i;
1239 DeleteCriticalSection(&ctx->tcp_state_mutex);
1240 for (i = 0; i < (int)POTR_MAX_PATH; i++)
1241 DeleteCriticalSection(&ctx->tcp_send_mutex[i]);
1242 DeleteCriticalSection(&ctx->recv_window_mutex);
1243 }
1244#endif /* _WIN32 */
1245 ctx_cleanup(ctx);
1246 return POTR_ERROR;
1247 }
1248 }
1249
1250 if (potr_connect_thread_start(ctx) != POTR_SUCCESS)
1251 {
1252 if (role == POTR_ROLE_SENDER
1253 || ctx->service.type == POTR_TYPE_TCP_BIDIR)
1254 {
1255 potr_send_queue_destroy(&ctx->send_queue);
1256 }
1257#ifndef _WIN32
1258 {
1259 int i;
1260 pthread_mutex_destroy(&ctx->tcp_state_mutex);
1261 pthread_cond_destroy(&ctx->tcp_state_cv);
1262 for (i = 0; i < (int)POTR_MAX_PATH; i++)
1263 pthread_mutex_destroy(&ctx->tcp_send_mutex[i]);
1264 pthread_mutex_destroy(&ctx->recv_window_mutex);
1265 }
1266#else /* _WIN32 */
1267 {
1268 int i;
1269 DeleteCriticalSection(&ctx->tcp_state_mutex);
1270 for (i = 0; i < (int)POTR_MAX_PATH; i++)
1271 DeleteCriticalSection(&ctx->tcp_send_mutex[i]);
1272 DeleteCriticalSection(&ctx->recv_window_mutex);
1273 }
1274#endif /* _WIN32 */
1275 ctx_cleanup(ctx);
1276 return POTR_ERROR;
1277 }
1278 }
1279 else
1280 {
1281 /* 非 TCP: 受信者の場合は受信スレッドのみ起動
1282 ただし UNICAST_BIDIR / UNICAST_BIDIR_N1 の RECEIVER は
1283 下の全スレッド起動ブロックで処理する */
1284 if (role == POTR_ROLE_RECEIVER
1285 && ctx->service.type != POTR_TYPE_UNICAST_BIDIR
1286 && ctx->service.type != POTR_TYPE_UNICAST_BIDIR_N1)
1287 {
1288 if (comm_recv_thread_start(ctx) != POTR_SUCCESS)
1289 {
1290 ctx_cleanup(ctx);
1291 return POTR_ERROR;
1292 }
1293 }
1294
1295 /* 非 TCP 送信者 / UNICAST_BIDIR / UNICAST_BIDIR_N1 受信者:
1296 送信キュー・送信スレッド・ヘルスチェックスレッド・受信スレッドを起動 */
1297 if (role == POTR_ROLE_SENDER
1298 || ctx->service.type == POTR_TYPE_UNICAST_BIDIR
1299 || ctx->service.type == POTR_TYPE_UNICAST_BIDIR_N1)
1300 {
1301 if (potr_send_queue_init(&ctx->send_queue,
1302 (size_t)ctx->global.send_queue_depth,
1303 ctx->global.max_payload) != POTR_SUCCESS)
1304 {
1305 ctx_cleanup(ctx);
1306 return POTR_ERROR;
1307 }
1308
1309 if (potr_send_thread_start(ctx) != POTR_SUCCESS)
1310 {
1311 potr_send_queue_destroy(&ctx->send_queue);
1312 ctx_cleanup(ctx);
1313 return POTR_ERROR;
1314 }
1315
1316 if (potr_health_thread_start(ctx) != POTR_SUCCESS)
1317 {
1318 potr_send_thread_stop(ctx);
1319 potr_send_queue_destroy(&ctx->send_queue);
1320 ctx_cleanup(ctx);
1321 return POTR_ERROR;
1322 }
1323
1324 if (comm_recv_thread_start(ctx) != POTR_SUCCESS)
1325 {
1326 potr_health_thread_stop(ctx);
1327 potr_send_thread_stop(ctx);
1328 potr_send_queue_destroy(&ctx->send_queue);
1329 ctx_cleanup(ctx);
1330 return POTR_ERROR;
1331 }
1332 }
1333 }
1334
1335 *handle = ctx;
1336 const char *role_str;
1337 if (role == POTR_ROLE_SENDER)
1338 {
1339 role_str = "SENDER";
1340 }
1341 else
1342 {
1343 role_str = "RECEIVER";
1344 }
1345 POTR_LOG(POTR_TRACE_INFO,
1346 "potrOpenService: service_id=%" PRId64 " role=%s encrypt=%s opened successfully",
1347 ctx->service.service_id,
1348 role_str,
1349 ctx->service.encrypt_enabled ? "ON" : "OFF");
1350 return POTR_SUCCESS;
1351}
compress.h
データ圧縮・解凍モジュールの内部ヘッダー。
POTR_COMPRESS_HEADER_SIZE
#define POTR_COMPRESS_HEADER_SIZE
圧縮ペイロード先頭に付加する元サイズフィールドのバイト数。
Definition compress.h:39
POTR_CRYPTO_TAG_SIZE
#define POTR_CRYPTO_TAG_SIZE
AES-256-GCM 認証タグサイズ (バイト)。暗号文の直後に付加する。
Definition porter_const.h:87
POTR_MAX_WINDOW_SIZE
#define POTR_MAX_WINDOW_SIZE
ウィンドウサイズの最大値 (パケット数)。window_size 設定値のバリデーション上限として使用する。
Definition porter_const.h:95
POTR_MAX_PATH
#define POTR_MAX_PATH
マルチパスの最大パス数。
Definition porter_const.h:100
POTR_MAX_PAYLOAD
#define POTR_MAX_PAYLOAD
ペイロードの最大長 (バイト)。UDP 最大ペイロード (65535 - IP20 - UDP8)。max_payload 設定値のバリデーション上限として使用する。
Definition porter_const.h:94
POTR_SUCCESS
#define POTR_SUCCESS
成功の戻り値を表す定数。
Definition porter_const.h:30
POTR_ERROR
#define POTR_ERROR
失敗の戻り値を表す定数。
Definition porter_const.h:31
packet.h
パケット構築・解析モジュールの内部ヘッダー。
PACKET_HEADER_SIZE
#define PACKET_HEADER_SIZE
パケットヘッダーの固定長 (バイト)。payload フィールドの開始オフセット。
Definition packet.h:23
porter.h
通信ライブラリ (動的リンク用) のヘッダーファイル。
POTR_API
#define POTR_API
呼び出し規約マクロ。
Definition porter.h:46
POTR_EXPORT
#define POTR_EXPORT
DLL エクスポート/インポート制御マクロ。
Definition porter.h:37
porter_const.h
通信ライブラリの定数ファイル。
POTR_TRACE_ERROR
@ POTR_TRACE_ERROR
エラー。操作の失敗を記録。TRACE_LV_ERROR (1) と同値。
Definition porter_type.h:335
POTR_TRACE_INFO
@ POTR_TRACE_INFO
情報。TRACE_LV_INFO (3) と同値。
Definition porter_type.h:337
POTR_TRACE_VERBOSE
@ POTR_TRACE_VERBOSE
詳細情報 (デバッグ)。TRACE_LV_VERBOSE (4) と同値。
Definition porter_type.h:338
PotrRecvCallback
void(* PotrRecvCallback)(int64_t service_id, PotrPeerId peer_id, PotrEvent event, const void *data, size_t len)
受信コールバック関数型 (全通信種別共通)。
Definition porter_type.h:308
POTR_TYPE_BROADCAST_RAW
@ POTR_TYPE_BROADCAST_RAW
1:N 通信 RAW モード (UDP ブロードキャスト)。
Definition porter_type.h:88
POTR_TYPE_BROADCAST
@ POTR_TYPE_BROADCAST
1:N 通信 (UDP ブロードキャスト)。
Definition porter_type.h:92
POTR_TYPE_UNICAST_BIDIR_N1
@ POTR_TYPE_UNICAST_BIDIR_N1
N:1 双方向通信 (UDP ユニキャスト)。
Definition porter_type.h:96
POTR_TYPE_UNICAST
@ POTR_TYPE_UNICAST
1:1 通信 (UDP ユニキャスト)。
Definition porter_type.h:90
POTR_TYPE_MULTICAST_RAW
@ POTR_TYPE_MULTICAST_RAW
1:N 通信 RAW モード (UDP マルチキャスト)。
Definition porter_type.h:87
POTR_TYPE_TCP_BIDIR
@ POTR_TYPE_TCP_BIDIR
TCP 双方向通信 (両端が potrSend 可)。
Definition porter_type.h:100
POTR_TYPE_TCP
@ POTR_TYPE_TCP
TCP ユニキャスト通信 (単方向: SENDER のみ potrSend 可)。
Definition porter_type.h:98
POTR_TYPE_UNICAST_RAW
@ POTR_TYPE_UNICAST_RAW
1:1 通信 RAW モード (UDP ユニキャスト)。
Definition porter_type.h:86
POTR_TYPE_UNICAST_BIDIR
@ POTR_TYPE_UNICAST_BIDIR
双方向 1:1 通信 (UDP ユニキャスト)。
Definition porter_type.h:94
POTR_TYPE_MULTICAST
@ POTR_TYPE_MULTICAST
1:N 通信 (UDP マルチキャスト)。
Definition porter_type.h:91
PotrHandle
struct PotrContext_ * PotrHandle
セッションハンドル。
Definition porter_type.h:238
PotrRole
PotrRole
役割種別。
Definition porter_type.h:114
POTR_ROLE_SENDER
@ POTR_ROLE_SENDER
送信者。
Definition porter_type.h:115
POTR_ROLE_RECEIVER
@ POTR_ROLE_RECEIVER
受信者。
Definition porter_type.h:116
potr_connect_thread_start
int potr_connect_thread_start(struct PotrContext_ *ctx)
TCP 接続管理スレッドを起動します (path 数分)。
Definition potrConnectThread.c:1084
potrConnectThread.h
TCP 接続管理スレッドの内部ヘッダー。
potrContext.h
セッションコンテキスト内部定義ヘッダー。
POTR_INVALID_SOCKET
#define POTR_INVALID_SOCKET
Definition potrContext.h:36
PotrSocket
int PotrSocket
Definition potrContext.h:32
potr_raw_base_type
static PotrType potr_raw_base_type(PotrType t)
RAW 系通信種別をベース通信種別に変換する (非 RAW 型はそのまま返す)。
Definition potrContext.h:62
potr_is_tcp_type
static int potr_is_tcp_type(PotrType t)
TCP 通信種別 (POTR_TYPE_TCP / POTR_TYPE_TCP_BIDIR) か判定する。
Definition potrContext.h:48
potr_health_thread_start
int potr_health_thread_start(struct PotrContext_ *ctx)
非 TCP ヘルスチェックスレッドを起動します。
Definition potrHealthThread.c:526
potr_health_thread_stop
int potr_health_thread_stop(struct PotrContext_ *ctx)
非 TCP ヘルスチェックスレッドを停止します。
Definition potrHealthThread.c:584
potrHealthThread.h
ヘルスチェックスレッド内部ヘッダー。
parse_ipv4_addr
int parse_ipv4_addr(const char *ip_str, struct in_addr *out_addr)
Definition potrIpAddr.c:28
resolve_ipv4_addr
int resolve_ipv4_addr(const char *host, struct in_addr *out_addr)
ホスト名または IPv4 アドレス文字列を struct in_addr に解決する。
Definition potrIpAddr.c:46
potrIpAddr.h
IPv4 アドレス変換ユーティリティ (内部用)。
potrLog.h
porter 内部ログマクロ定義ヘッダー。
POTR_LOG
#define POTR_LOG(level,...)
porter 内部ログ出力マクロ。
Definition potrLog.h:68
generate_session
static void generate_session(struct PotrContext_ *ctx)
Definition potrOpenService.c:85
open_socket_broadcast
static PotrSocket open_socket_broadcast(uint16_t src_port, uint16_t dst_port, struct in_addr src_if, int is_receiver)
Definition potrOpenService.c:211
open_socket_tcp_receiver
static int open_socket_tcp_receiver(struct PotrContext_ *ctx, int path_idx)
Definition potrOpenService.c:343
open_socket_multicast
static PotrSocket open_socket_multicast(const PotrServiceDef *def, struct in_addr src_if, int is_receiver)
Definition potrOpenService.c:115
open_socket_tcp_sender
static int open_socket_tcp_sender(struct PotrContext_ *ctx, int path_idx)
Definition potrOpenService.c:417
open_socket_unicast
static PotrSocket open_socket_unicast(struct in_addr bind_addr, uint16_t port)
Definition potrOpenService.c:47
ctx_cleanup
static void ctx_cleanup(struct PotrContext_ *ctx)
Definition potrOpenService.c:293
potrOpenService
POTR_EXPORT int POTR_API potrOpenService(const PotrGlobalConfig *global, const PotrServiceDef *service, PotrRole role, PotrRecvCallback callback, PotrHandle *handle)
設定構造体から指定サービスを開きます。
Definition potrOpenService.c:445
cleanup_sockets
static void cleanup_sockets(struct PotrContext_ *ctx)
Definition potrOpenService.c:274
peer_table_destroy
void peer_table_destroy(struct PotrContext_ *ctx)
ピアテーブルを破棄する。
Definition potrPeerTable.c:228
peer_table_init
int peer_table_init(struct PotrContext_ *ctx)
ピアテーブルを初期化する。
Definition potrPeerTable.c:197
potrPeerTable.h
N:1 モード用ピアテーブル管理モジュールの内部ヘッダー。
comm_recv_thread_start
int comm_recv_thread_start(struct PotrContext_ *ctx)
非 TCP 受信スレッドを起動します。
Definition potrRecvThread.c:2980
potrRecvThread.h
受信スレッド内部ヘッダー。
potr_send_queue_init
int potr_send_queue_init(PotrSendQueue *q, size_t depth, uint16_t max_payload)
Definition potrSendQueue.c:51
potr_send_queue_destroy
void potr_send_queue_destroy(PotrSendQueue *q)
Definition potrSendQueue.c:87
potrSendQueue.h
非同期送信キューの型定義と操作関数。
potr_send_thread_start
int potr_send_thread_start(struct PotrContext_ *ctx)
Definition potrSendThread.c:710
potr_send_thread_stop
void potr_send_thread_stop(struct PotrContext_ *ctx)
Definition potrSendThread.c:737
potrSendThread.h
非同期送信スレッドの内部ヘッダー。
PotrContext_
セッションコンテキスト構造体。PotrHandle の実体。
Definition potrContext.h:156
PotrContext_::session_id
uint32_t session_id
自セッション識別子 (乱数)。
Definition potrContext.h:183
PotrContext_::callback
PotrRecvCallback callback
受信コールバック。
Definition potrContext.h:157
PotrContext_::global
PotrGlobalConfig global
グローバル設定。
Definition potrContext.h:163
PotrContext_::dst_addr_resolved
struct in_addr dst_addr_resolved[POTR_MAX_PATH]
解決済み宛先 IPv4 アドレス (unicast のみ)。
Definition potrContext.h:179
PotrContext_::compress_buf_size
size_t compress_buf_size
compress_buf のサイズ (バイト)。
Definition potrContext.h:210
PotrContext_::frag_buf
uint8_t * frag_buf
フラグメント結合バッファ (動的確保。max_message_size バイト)。
Definition potrContext.h:208
PotrContext_::health_wakeup
PotrCondVar health_wakeup[POTR_MAX_PATH]
ヘルスチェックスレッドを即時起床させる条件変数 (path ごと)。
Definition potrContext.h:161
PotrContext_::src_addr_resolved
struct in_addr src_addr_resolved[POTR_MAX_PATH]
解決済み送信元 IPv4 アドレス。
Definition potrContext.h:178
PotrContext_::session_tv_nsec
int32_t session_tv_nsec
自セッション開始時刻 ナノ秒部。
Definition potrContext.h:185
PotrContext_::role
PotrRole role
役割 (POTR_ROLE_SENDER / POTR_ROLE_RECEIVER)。
Definition potrContext.h:175
PotrContext_::crypto_buf
uint8_t * crypto_buf
暗号化・復号用一時バッファ (動的確保)。
Definition potrContext.h:211
PotrContext_::send_queue
PotrSendQueue send_queue
非同期送信キュー。
Definition potrContext.h:234
PotrContext_::compress_buf
uint8_t * compress_buf
圧縮・解凍用一時バッファ (動的確保)。
Definition potrContext.h:209
PotrContext_::recv_buf
uint8_t * recv_buf
受信バッファ / 再送 wire 組立バッファ (動的確保。PACKET_HEADER_SIZE + max_payload バイト)。
Definition potrContext.h:213
PotrContext_::send_wire_buf
uint8_t * send_wire_buf
送信 wire 組立バッファ (動的確保。PACKET_HEADER_SIZE + max_payload バイト)。送信スレッドのみ使用。
Definition potrContext.h:214
PotrContext_::tcp_send_mutex
PotrMutex tcp_send_mutex[POTR_MAX_PATH]
TCP send() 排他制御 (path ごと)。送信スレッド・ヘルスチェックスレッド・recv スレッド競合防止。
Definition potrContext.h:250
PotrContext_::tcp_state_mutex
PotrMutex tcp_state_mutex
tcp_state_cv 保護用ミューテックス。tcp_active_paths のカウンタ更新も保護。
Definition potrContext.h:260
PotrContext_::tcp_listen_sock
PotrSocket tcp_listen_sock[POTR_MAX_PATH]
RECEIVER: listen ソケット (path ごと)。
Definition potrContext.h:246
PotrContext_::is_multi_peer
int is_multi_peer
1: N:1 モード (src_addr/src_port 省略), 0: 1:1 モード。
Definition potrContext.h:237
PotrContext_::service
PotrServiceDef service
サービス定義。
Definition potrContext.h:162
PotrContext_::health_mutex
PotrMutex health_mutex[POTR_MAX_PATH]
ヘルスチェックスレッド停止用ミューテックス (path ごと)。
Definition potrContext.h:160
PotrContext_::crypto_buf_size
size_t crypto_buf_size
crypto_buf のサイズ (バイト)。
Definition potrContext.h:212
PotrContext_::sock
PotrSocket sock[POTR_MAX_PATH]
各パスの UDP ソケット。
Definition potrContext.h:169
PotrContext_::send_window
PotrWindow send_window
送信バッファ (過去 N パケット保持。NACK 再送・REJECT 判定に使用)。
Definition potrContext.h:164
PotrContext_::max_peers
int max_peers
ピアテーブルサイズ (service.max_peers から取得)。
Definition potrContext.h:240
PotrContext_::tcp_state_cv
PotrCondVar tcp_state_cv
切断通知・reconnect sleep の中断用条件変数。
Definition potrContext.h:261
PotrContext_::recv_window_mutex
PotrMutex recv_window_mutex
recv_window 保護用ミューテックス。
Definition potrContext.h:253
PotrContext_::dest_addr
struct sockaddr_in dest_addr[POTR_MAX_PATH]
送信先ソケットアドレス (送信者が sendto に使用)。
Definition potrContext.h:180
PotrContext_::recv_window
PotrWindow recv_window
受信ウィンドウ (順序整列・欠番検出)。
Definition potrContext.h:166
PotrContext_::session_tv_sec
int64_t session_tv_sec
自セッション開始時刻 秒部。
Definition potrContext.h:184
PotrContext_::n_path
int n_path
有効パス数。
Definition potrContext.h:170
PotrContext_::peers
PotrPeerContext * peers
ピアテーブル (動的確保。max_peers エントリ)。
Definition potrContext.h:239
PotrContext_::tcp_conn_fd
PotrSocket tcp_conn_fd[POTR_MAX_PATH]
アクティブ TCP 接続 fd (path ごと)。
Definition potrContext.h:247
PotrGlobalConfig
グローバル設定。
Definition porter_type.h:177
PotrGlobalConfig::tcp_health_interval_ms
uint32_t tcp_health_interval_ms
TCP 通信種別の PING 送信間隔 (ミリ秒)。DATA 送信頻度に関わらず定期的に PING を送信。0 = 無効。設定ファイルキー: tcp_health_interval_ms。
Definition porter_type.h:185
PotrGlobalConfig::window_size
uint16_t window_size
スライディングウィンドウサイズ (パケット数)。
Definition porter_type.h:178
PotrGlobalConfig::send_queue_depth
uint32_t send_queue_depth
非同期送信キューの最大エントリ数。デフォルト: POTR_SEND_QUEUE_DEPTH。
Definition porter_type.h:184
PotrGlobalConfig::tcp_health_timeout_ms
uint32_t tcp_health_timeout_ms
TCP 通信種別の PING 応答待機タイムアウト (ミリ秒)。SENDER 側で使用。0 = 無効。設定ファイルキー: tcp_health_timeout_ms。
Definition porter_type.h:186
PotrGlobalConfig::max_payload
uint16_t max_payload
最大ペイロード長 (バイト)。
Definition porter_type.h:179
PotrGlobalConfig::health_interval_ms
uint32_t health_interval_ms
UDP 通信種別の PING 送信間隔 (ミリ秒)。最終 DATA/PING 送信から本値が経過したら PING 送信。0 = 無効。設定ファイルキー: udp_health_interval_ms。
Definition porter_type.h:180
PotrGlobalConfig::health_timeout_ms
uint32_t health_timeout_ms
UDP 通信種別の受信タイムアウト (ミリ秒)。RECEIVER 側で使用。0 = 無効。設定ファイルキー: udp_health_timeout_ms。
Definition porter_type.h:181
PotrGlobalConfig::max_message_size
uint32_t max_message_size
1 回の potrSend で送信できる最大メッセージ長 (バイト)。デフォルト: POTR_MAX_MESSAGE_SIZE。
Definition porter_type.h:183
PotrServiceDef
サービス定義。
Definition porter_type.h:130
PotrServiceDef::src_port
uint16_t src_port
送信者の送信元 bind ポート番号。0 = OS 自動選定。(全通信種別で省略可)
Definition porter_type.h:138
PotrServiceDef::health_interval_ms
uint32_t health_interval_ms
グローバルの udp/tcp_health_interval_ms をサービス単位で上書きする。0 = グローバル値を使用。
Definition porter_type.h:160
PotrServiceDef::encrypt_enabled
int encrypt_enabled
非 0 のとき暗号化有効。設定ファイルに有効な encrypt_key が存在するときに 1 に設定される。
Definition porter_type.h:154
PotrServiceDef::dst_port
uint16_t dst_port
宛先ポート番号。サービスの識別子。受信者の bind ポート / 送信者の送信先ポート。(全通信種別で必須)
Definition porter_type.h:135
PotrServiceDef::src_addr
char src_addr[POTR_MAX_PATH][POTR_MAX_ADDR_LEN]
送信元アドレス [0]=src_addr1 〜 [3]=src_addr4。送信者は bind / 送信インターフェース、受信者は送信元フィルタ。(全通信種別で必須)
Definition porter_type.h:149
PotrServiceDef::multicast_group
char multicast_group[POTR_MAX_ADDR_LEN]
マルチキャストグループアドレス。(multicast のみ)
Definition porter_type.h:145
PotrServiceDef::type
PotrType type
通信種別。
Definition porter_type.h:132
PotrServiceDef::service_id
int64_t service_id
サービス ID。
Definition porter_type.h:131
PotrServiceDef::dst_addr
char dst_addr[POTR_MAX_PATH][POTR_MAX_ADDR_LEN]
宛先アドレス [0]=dst_addr1 〜 [3]=dst_addr4。送信者は送信先、受信者は bind アドレス。(unicast のみ)
Definition porter_type.h:150
PotrServiceDef::max_peers
uint32_t max_peers
N:1 モード時の最大同時接続ピア数。省略時: 1024。1:1 モードでは無視される。
Definition porter_type.h:157
PotrServiceDef::broadcast_addr
char broadcast_addr[POTR_MAX_ADDR_LEN]
ブロードキャスト宛先アドレス。省略時は 255.255.255.255。(broadcast のみ)
Definition porter_type.h:146
PotrServiceDef::health_timeout_ms
uint32_t health_timeout_ms
グローバルの udp/tcp_health_timeout_ms をサービス単位で上書きする。0 = グローバル値を使用。
Definition porter_type.h:161
window_init
int window_init(PotrWindow *win, uint32_t initial_seq, uint16_t window_size, uint16_t max_payload)
ウィンドウを初期化します。
Definition window.c:42
window_destroy
void window_destroy(PotrWindow *win)
ウィンドウが保持する動的確保バッファを解放します。
Definition window.c:110
window.h
スライディングウィンドウ管理モジュールの内部ヘッダー。