source: benchmark/io/http/main.cfa@ d1b70d4

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

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