source: benchmark/io/http/main.cfa@ 0edd11a

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 0edd11a was 8c43d05, checked in by Thierry Delisle <tdelisle@…>, 5 years ago

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