source: benchmark/io/http/main.cfa@ 1adab3e

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

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