source: benchmark/io/http/main.cfa@ 348f81d5

ADT arm-eh ast-experimental enum forall-pointer-decay jacob/cs343-translation new-ast-unique-expr pthread-emulation qualifiedEnum
Last change on this file since 348f81d5 was 348f81d5, checked in by Thierry Delisle <tdelisle@…>, 5 years ago

Added option to isolate processor so everything has it's own cluster
  • Property mode set to 100644
File size: 7.6 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 main(StatsPrinter & this) {
41 LOOP: for() {
42 waitfor( ^?{} : this) {
43 break LOOP;
44 }
45 or else {}
46
47 sleep(10`s);
48
49 print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
50 }
51}
52
53//=============================================================================================
54// Globals
55//=============================================================================================
56// struct ServerProc {
57// processor self;
58// };
59
60// void ?{}( ServerProc & this ) {
61// /* paranoid */ assert( options.clopts.instance != 0p );
62// (this.self){ "Benchmark Processor", *options.clopts.instance };
63
64// #if !defined(__CFA_NO_STATISTICS__)
65// if( options.clopts.procstats ) {
66// print_stats_at_exit( this.self, options.clopts.instance->print_stats );
67// }
68// if( options.clopts.viewhalts ) {
69// print_halts( this.self );
70// }
71// #endif
72// }
73
74// struct ServerIoContext {
75// io_context self;
76// };
77
78// void ?{}( ServerIoContext & this ) {
79// /* paranoid */ assert( options.clopts.instance != 0p );
80// (this.self){ *options.clopts.instance };
81// }
82
83struct ServerCluster {
84 cluster self;
85 processor * procs;
86 io_context * ctxs;
87 StatsPrinter * prnt;
88
89};
90
91void ?{}( ServerCluster & this ) {
92 (this.self){ "Server Cluster", options.clopts.params };
93
94 this.procs = alloc(options.clopts.nprocs);
95 for(i; options.clopts.nprocs) {
96 (this.procs[i]){ "Benchmark Processor", this.self };
97
98 #if !defined(__CFA_NO_STATISTICS__)
99 if( options.clopts.procstats ) {
100 print_stats_at_exit( *this.procs, this.self.print_stats );
101 }
102 if( options.clopts.viewhalts ) {
103 print_halts( *this.procs );
104 }
105 #endif
106 }
107
108 this.ctxs = alloc(options.clopts.nprocs);
109 for(i; options.clopts.nprocs) {
110 (this.ctxs[i]){ this.self };
111 }
112
113 this.prnt = alloc(options.clopts.nprocs);
114 for(i; options.clopts.nprocs) {
115 (this.prnt[i]){ this.self };
116 }
117
118 #if !defined(__CFA_NO_STATISTICS__)
119 print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO );
120 #endif
121
122 options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
123 options.clopts.cltr_cnt++;
124}
125
126void ^?{}( ServerCluster & this ) {
127 for(i; options.clopts.nprocs) {
128 ^(this.prnt[i]){};
129 }
130 free(this.prnt);
131
132 for(i; options.clopts.nprocs) {
133 ^(this.ctxs[i]){};
134 }
135 free(this.ctxs);
136
137 for(i; options.clopts.nprocs) {
138 ^(this.procs[i]){};
139 }
140 free(this.procs);
141
142 ^(this.self){};
143}
144
145extern void init_protocol(void);
146extern void deinit_protocol(void);
147
148//=============================================================================================
149// Main
150//============================================================================================='
151int main( int argc, char * argv[] ) {
152 __sighandler_t s = 1p;
153 signal(SIGPIPE, s);
154
155 //===================
156 // Parse args
157 parse_options(argc, argv);
158
159 //===================
160 // Open Files
161 if( options.file_cache.path ) {
162 sout | "Filling cache from" | options.file_cache.path;
163 fill_cache( options.file_cache.path );
164 }
165
166 //===================
167 // Open Socket
168 sout | getpid() | ": Listening on port" | options.socket.port;
169 int server_fd = socket(AF_INET, SOCK_STREAM, 0);
170 if(server_fd < 0) {
171 abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
172 }
173
174 int ret = 0;
175 struct sockaddr_in address;
176 int addrlen = sizeof(address);
177 memset( (char *)&address, '\0' );
178 address.sin_family = AF_INET;
179 address.sin_addr.s_addr = htonl(INADDR_ANY);
180 address.sin_port = htons( options.socket.port );
181
182 int waited = 0;
183 for() {
184 ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
185 if(ret < 0) {
186 if(errno == EADDRINUSE) {
187 if(waited == 0) {
188 sout | "Waiting for port";
189 } else {
190 sout | "\r" | waited | nonl;
191 flush( sout );
192 }
193 waited ++;
194 sleep( 1`s );
195 continue;
196 }
197 abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
198 }
199 break;
200 }
201
202 ret = listen( server_fd, options.socket.backlog );
203 if(ret < 0) {
204 abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
205 }
206
207 //===================
208 // Run Server Cluster
209 {
210 int pipe_cnt = options.clopts.nworkers * 2;
211 int pipe_off;
212 int * fds;
213 [fds, pipe_off] = filefds( pipe_cnt );
214 for(i; 0 ~ pipe_cnt ~ 2) {
215 int ret = pipe(&fds[pipe_off + i]);
216 if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
217 }
218
219 // if(options.file_cache.path && options.file_cache.fixed_fds) {
220 // register_fixed_files(cl, fds, pipe_off);
221 // }
222
223 {
224 ServerCluster cl[options.clopts.nclusters];
225
226 init_protocol();
227 {
228 Worker workers[options.clopts.nworkers];
229 for(i; options.clopts.nworkers) {
230 // if( options.file_cache.fixed_fds ) {
231 // workers[i].pipe[0] = pipe_off + (i * 2) + 0;
232 // workers[i].pipe[1] = pipe_off + (i * 2) + 1;
233 // }
234 // else
235 {
236 workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
237 workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
238 workers[i].sockfd = server_fd;
239 workers[i].addr = (struct sockaddr *)&address;
240 workers[i].addrlen = (socklen_t*)&addrlen;
241 workers[i].flags = 0;
242 }
243 unpark( workers[i] );
244 }
245 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters";
246 for(i; options.clopts.nclusters) {
247 sout | options.clopts.thrd_cnt[i] | nonl;
248 }
249 sout | nl;
250 {
251 char buffer[128];
252 while(int ret = cfa_read(0, buffer, 128, 0); ret != 0) {
253 if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
254 }
255
256 sout | "Shutdown received";
257 }
258
259 sout | "Notifying connections..." | nonl; flush( sout );
260 for(i; options.clopts.nworkers) {
261 workers[i].done = true;
262 }
263 sout | "done";
264
265 sout | "Shutting down socket..." | nonl; flush( sout );
266 int ret = shutdown( server_fd, SHUT_RD );
267 if( ret < 0 ) {
268 abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
269 }
270 sout | "done";
271
272 //===================
273 // Close Socket
274 sout | "Closing Socket..." | nonl; flush( sout );
275 ret = close( server_fd );
276 if(ret < 0) {
277 abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
278 }
279 sout | "done";
280
281 sout | "Stopping connection threads..." | nonl; flush( sout );
282 }
283 sout | "done";
284
285 sout | "Stopping protocol threads..." | nonl; flush( sout );
286 deinit_protocol();
287 sout | "done";
288
289 sout | "Stopping processors/clusters..." | nonl; flush( sout );
290 }
291 sout | "done";
292
293 sout | "Closing splice fds..." | nonl; flush( sout );
294 for(i; pipe_cnt) {
295 ret = close( fds[pipe_off + i] );
296 if(ret < 0) {
297 abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
298 }
299 }
300 free(fds);
301 sout | "done";
302
303 sout | "Stopping processors..." | nonl; flush( sout );
304 }
305 sout | "done";
306
307 //===================
308 // Close Files
309 if( options.file_cache.path ) {
310 sout | "Closing open files..." | nonl; flush( sout );
311 close_cache();
312 sout | "done";
313 }
314}
Note: See TracBrowser for help on using the repository browser.