Context Navigation

main.cfa @ cb921d4

ADTast-experimentalenumforall-pointer-decaypthread-emulationqualifiedEnum

Last change on this file since cb921d4 was 3eb540fb, checked in by Thierry Delisle <tdelisle@…>, 3 years ago
Modified httpforall not to use VLA of threads.
Property mode set to `100644`
File size: 6.9 KB

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

Note: See TracBrowser for help on using the repository browser.

Download in other formats:

Original Format