source: benchmark/io/http/main.cfa@ 4f762d3

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 4f762d3 was 4f762d3, checked in by Thierry Delisle <tdelisle@…>, 5 years ago

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