Mercurial

comparison seobeo/os/s_macos_edge.c @ 62:ea9ef388ab97

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[Seobeo] Fixed issues with epoll or kqeue in different threads. Initizlied the event looop inside of the thread itself.
author June Park <parkjune1995@gmail.com>
date Tue, 23 Dec 2025 11:48:11 -0800
parents c0f6c8c7829f
children ecb6ee6a22c3
comparison
equal deleted inserted replaced
61:9df5587cf23b 62:ea9ef388ab97
4 4
5 void *Seobeo_Web_Edge_Worker(void *vargs) 5 void *Seobeo_Web_Edge_Worker(void *vargs)
6 { 6 {
7 WorkerArgs *args = vargs; 7 WorkerArgs *args = vargs;
8 struct kevent evlist[64]; 8 struct kevent evlist[64];
9
10 // Each thread creates its own kqueue to avoid race conditions
11 int kq = kqueue();
12 if (kq < 0) {
13 perror("kqueue");
14 return NULL;
15 }
16
17 // Add server socket to this thread's kqueue
18 struct kevent kev = {
19 .ident = args->srv->socket,
20 .filter = EVFILT_READ,
21 .flags = EV_ADD,
22 .udata = args->srv
23 };
24 kevent(kq, &kev, 1, NULL, 0, NULL);
25
9 while (1) 26 while (1)
10 { 27 {
11 int ne = kevent(args->evfd, NULL, 0, evlist, 64, NULL); 28 int ne = kevent(kq, NULL, 0, evlist, 64, NULL);
12 if (ne < 0) continue; 29 if (ne < 0) {
30 if (errno == EINTR) continue;
31 perror("kevent");
32 continue;
33 }
13 34
14 for (int i = 0; i < ne; i++) 35 for (int i = 0; i < ne; i++)
15 { 36 {
16 Seobeo_PHandle h = evlist[i].udata; 37 Seobeo_PHandle h = evlist[i].udata;
17 if (h == args->srv) 38 if (h == args->srv)
18 { 39 {
19 Seobeo_PHandle cli = 40 // Accept new connections in a loop (for edge-triggered behavior)
20 Seobeo_Stream_Handle_Server_Accept(args->srv); 41 while (1) {
42 Seobeo_PHandle cli = Seobeo_Stream_Handle_Server_Accept(args->srv);
43 if (!cli) break;
21 44
22 if (!cli) continue; 45 struct kevent client_kev = {
23 struct kevent kev = { 46 .ident = cli->socket,
24 .ident = cli->socket, 47 .filter = EVFILT_READ,
48 .flags = EV_ADD | EV_ONESHOT,
49 .udata = cli
50 };
51 kevent(kq, &client_kev, 1, NULL, 0, NULL);
52 }
53 } else {
54 // Remove from kqueue first
55 struct kevent del_kev = {
56 .ident = h->socket,
25 .filter = EVFILT_READ, 57 .filter = EVFILT_READ,
26 .flags = EV_ADD | EV_ONESHOT, 58 .flags = EV_DELETE,
27 .udata = cli
28 }; 59 };
29 kevent(args->evfd, &kev, 1, NULL, 0, NULL); 60 kevent(kq, &del_kev, 1, NULL, 0, NULL);
30 } else { 61
31 if (h != args->srv) { 62 // Handle request (this function destroys the handle internally)
32 struct kevent kev = { 63 Seobeo_Web_HandleClientRequest(h, args->cache);
33 .ident = h->socket,
34 .filter = EVFILT_READ,
35 .flags = EV_DELETE,
36 };
37 kevent(args->evfd, &kev, 1, NULL, 0, NULL); // Remove from kqueue first
38
39 Seobeo_Web_HandleClientRequest(h, args->cache); // this frees
40 }
41 } 64 }
42 } 65 }
43 } 66 }
67
68 close(kq);
44 return NULL; 69 return NULL;
45 } 70 }
46 71
47 void Seobeo_Web_Edge( 72 void Seobeo_Web_Edge(
48 Seobeo_PHandle p_server_handle, 73 Seobeo_PHandle p_server_handle,
49 int thread_count, 74 int thread_count,
50 Dowa_PHashMap p_html_cache) 75 Dowa_PHashMap p_html_cache)
51 { 76 {
52 int kq = kqueue();
53 struct kevent kev = {
54 .ident = p_server_handle->socket,
55 .filter = EVFILT_READ,
56 .flags = EV_ADD,
57 .udata = p_server_handle
58 };
59 kevent(kq, &kev, 1, NULL, 0, NULL);
60
61 pthread_attr_t attr; 77 pthread_attr_t attr;
62 pthread_attr_init(&attr); 78 pthread_attr_init(&attr);
63 pthread_attr_setstacksize(&attr, 5 * 1024 * 1024); // 5 MB 79 pthread_attr_setstacksize(&attr, 5 * 1024 * 1024); // 5 MB
64 80
65 pthread_t threads[thread_count]; 81 pthread_t threads[thread_count];
66 for (int i = 0; i < thread_count; i++) 82 for (int i = 0; i < thread_count; i++)
67 { 83 {
68 WorkerArgs *args = malloc(sizeof(WorkerArgs)); 84 WorkerArgs *args = malloc(sizeof(WorkerArgs));
69 *args = (WorkerArgs){ p_server_handle, p_html_cache, kq }; 85 *args = (WorkerArgs){ p_server_handle, p_html_cache };
70 86
71 pthread_create(&threads[i], NULL, Seobeo_Web_Edge_Worker, args); 87 pthread_create(&threads[i], &attr, Seobeo_Web_Edge_Worker, args);
72 } 88 }
73 for (int i = 0; i < thread_count; i++) 89 for (int i = 0; i < thread_count; i++)
74 { 90 {
75 pthread_join(threads[i], NULL); 91 pthread_join(threads[i], NULL);
76 } 92 }
93
94 pthread_attr_destroy(&attr);
77 return; 95 return;
78 } 96 }