source: benchmark/io/http/main.cfa@ 1387ea0

ADT ast-experimental enum forall-pointer-decay pthread-emulation qualifiedEnum
Last change on this file since 1387ea0 was 3eb540fb, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

Modified httpforall not to use VLA of threads.
  • Property mode set to 100644
File size: 6.9 KB
Line 
1#define __USE_GNU
2
3#include <errno.h>
4#include <stdio.h>
5#include <string.h>
6#include <unistd.h>
7extern "C" {
8 #include <signal.h>
9 #include <sys/socket.h>
10 #include <netinet/in.h>
11}
12
13#include <fstream.hfa>
14#include <kernel.hfa>
15#include <iofwd.hfa>
16#include <stats.hfa>
17#include <time.hfa>
18#include <thread.hfa>
19
20#include "filecache.hfa"
21#include "options.hfa"
22#include "worker.hfa"
23
24extern void register_fixed_files( cluster &, int *, unsigned count );
25
26Duration default_preemption() {
27 return 0;
28}
29
30//=============================================================================================
31// Stats Printer
32//============================================================================================='
33
34thread StatsPrinter {};
35
36void ?{}( StatsPrinter & this, cluster & cl ) {
37 ((thread&)this){ "Stats Printer Thread", cl };
38}
39
40void ^?{}( StatsPrinter & mutex this ) {}
41
42void main(StatsPrinter & this) {
43 LOOP: for() {
44 waitfor( ^?{} : this) {
45 break LOOP;
46 }
47 or else {}
48
49 sleep(10`s);
50
51 print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
52 }
53}
54
55//=============================================================================================
56// Globals
57//=============================================================================================
58struct ServerCluster {
59 cluster self;
60 processor * procs;
61 // io_context * ctxs;
62 StatsPrinter * prnt;
63
64};
65
66void ?{}( ServerCluster & this ) {
67 (this.self){ "Server Cluster", options.clopts.params };
68
69 this.procs = alloc(options.clopts.nprocs);
70 for(i; options.clopts.nprocs) {
71 (this.procs[i]){ "Benchmark Processor", this.self };
72
73 #if !defined(__CFA_NO_STATISTICS__)
74 if( options.clopts.procstats ) {
75 print_stats_at_exit( *this.procs, this.self.print_stats );
76 }
77 if( options.clopts.viewhalts ) {
78 print_halts( *this.procs );
79 }
80 #endif
81 }
82
83 if(options.stats) {
84 this.prnt = alloc();
85 (*this.prnt){ this.self };
86 } else {
87 this.prnt = 0p;
88 }
89
90 #if !defined(__CFA_NO_STATISTICS__)
91 print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO );
92 #endif
93
94 options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
95 options.clopts.cltr_cnt++;
96}
97
98void ^?{}( ServerCluster & this ) {
99 delete(this.prnt);
100
101 for(i; options.clopts.nprocs) {
102 ^(this.procs[i]){};
103 }
104 free(this.procs);
105
106 ^(this.self){};
107}
108
109extern void init_protocol(void);
110extern void deinit_protocol(void);
111
112//=============================================================================================
113// Main
114//============================================================================================='
115int main( int argc, char * argv[] ) {
116 __sighandler_t s = 1p;
117 signal(SIGPIPE, s);
118
119 //===================
120 // Parse args
121 parse_options(argc, argv);
122
123 //===================
124 // Open Files
125 if( options.file_cache.path ) {
126 sout | "Filling cache from" | options.file_cache.path;
127 fill_cache( options.file_cache.path );
128 }
129
130 //===================
131 // Open Socket
132 sout | getpid() | ": Listening on port" | options.socket.port;
133 int server_fd = socket(AF_INET, SOCK_STREAM, 0);
134 if(server_fd < 0) {
135 abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
136 }
137
138 int ret = 0;
139 struct sockaddr_in address;
140 int addrlen = sizeof(address);
141 memset( (char *)&address, '\0' );
142 address.sin_family = AF_INET;
143 address.sin_addr.s_addr = htonl(INADDR_ANY);
144 address.sin_port = htons( options.socket.port );
145
146 int waited = 0;
147 for() {
148 ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
149 if(ret < 0) {
150 if(errno == EADDRINUSE) {
151 if(waited == 0) {
152 sout | "Waiting for port";
153 } else {
154 sout | "\r" | waited | nonl;
155 flush( sout );
156 }
157 waited ++;
158 sleep( 1`s );
159 continue;
160 }
161 abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
162 }
163 break;
164 }
165
166 ret = listen( server_fd, options.socket.backlog );
167 if(ret < 0) {
168 abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
169 }
170
171 //===================
172 // Run Server Cluster
173 {
174 int pipe_cnt = options.clopts.nworkers * 2;
175 int pipe_off;
176 int * fds;
177 [fds, pipe_off] = filefds( pipe_cnt );
178 for(i; 0 ~ pipe_cnt ~ 2) {
179 int ret = pipe(&fds[pipe_off + i]);
180 if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
181 }
182
183 // if(options.file_cache.path && options.file_cache.fixed_fds) {
184 // register_fixed_files(cl, fds, pipe_off);
185 // }
186
187 {
188 ServerCluster cl[options.clopts.nclusters];
189
190 init_protocol();
191 {
192 Worker * workers = anew(options.clopts.nworkers);
193 for(i; options.clopts.nworkers) {
194 // if( options.file_cache.fixed_fds ) {
195 // workers[i].pipe[0] = pipe_off + (i * 2) + 0;
196 // workers[i].pipe[1] = pipe_off + (i * 2) + 1;
197 // }
198 // else
199 {
200 workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
201 workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
202 workers[i].sockfd = server_fd;
203 workers[i].addr = (struct sockaddr *)&address;
204 workers[i].addrlen = (socklen_t*)&addrlen;
205 workers[i].flags = 0;
206 }
207 unpark( workers[i] );
208 }
209 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters";
210 for(i; options.clopts.nclusters) {
211 sout | options.clopts.thrd_cnt[i] | nonl;
212 }
213 sout | nl;
214 if(!options.interactive) park();
215 {
216 char buffer[128];
217 for() {
218 int ret = cfa_read(0, buffer, 128, 0);
219 if(ret == 0) break;
220 if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
221 sout | "User wrote '" | "" | nonl;
222 write(sout, buffer, ret - 1);
223 sout | "'";
224 }
225
226 sout | "Shutdown received";
227 }
228
229 sout | "Notifying connections..." | nonl; flush( sout );
230 for(i; options.clopts.nworkers) {
231 workers[i].done = true;
232 }
233 sout | "done";
234
235 sout | "Shutting down socket..." | nonl; flush( sout );
236 int ret = shutdown( server_fd, SHUT_RD );
237 if( ret < 0 ) {
238 abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
239 }
240 sout | "done";
241
242 //===================
243 // Close Socket
244 sout | "Closing Socket..." | nonl; flush( sout );
245 ret = close( server_fd );
246 if(ret < 0) {
247 abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
248 }
249 sout | "done";
250
251 sout | "Stopping connection threads..." | nonl; flush( sout );
252 adelete(workers);
253 }
254 sout | "done";
255
256 sout | "Stopping protocol threads..." | nonl; flush( sout );
257 deinit_protocol();
258 sout | "done";
259
260 sout | "Stopping processors/clusters..." | nonl; flush( sout );
261 }
262 sout | "done";
263
264 sout | "Closing splice fds..." | nonl; flush( sout );
265 for(i; pipe_cnt) {
266 ret = close( fds[pipe_off + i] );
267 if(ret < 0) {
268 abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
269 }
270 }
271 free(fds);
272 sout | "done";
273
274 sout | "Stopping processors..." | nonl; flush( sout );
275 }
276 sout | "done";
277
278 //===================
279 // Close Files
280 if( options.file_cache.path ) {
281 sout | "Closing open files..." | nonl; flush( sout );
282 close_cache();
283 sout | "done";
284 }
285}
Note: See TracBrowser for help on using the repository browser.