93 	void *(*run)(void *svc);
440 static int rs_notify_svc(struct rs_svc *svc, struct rsocket *rs, int cmd)  in rs_notify_svc()  argument
446 	if (!svc->cnt) {  in rs_notify_svc()
447 		ret = socketpair(AF_UNIX, SOCK_STREAM, 0, svc->sock);  in rs_notify_svc()
451 		ret = pthread_create(&svc->id, NULL, svc->run, svc);  in rs_notify_svc()
461 	write_all(svc->sock[0], &msg, sizeof msg);  in rs_notify_svc()
462 	read_all(svc->sock[0], &msg, sizeof msg);  in rs_notify_svc()
464 	if (svc->cnt)  in rs_notify_svc()
467 	pthread_join(svc->id, NULL);  in rs_notify_svc()
469 	close(svc->sock[0]);  in rs_notify_svc()
470 	close(svc->sock[1]);  in rs_notify_svc()
3944 static int rs_svc_grow_sets(struct rs_svc *svc, int grow_size)  in rs_svc_grow_sets()  argument
3949 	set = calloc(svc->size + grow_size, sizeof(*rss) + svc->context_size);  in rs_svc_grow_sets()
3953 	svc->size += grow_size;  in rs_svc_grow_sets()
3955 	contexts = set + sizeof(*rss) * svc->size;  in rs_svc_grow_sets()
3956 	if (svc->cnt) {  in rs_svc_grow_sets()
3957 		memcpy(rss, svc->rss, sizeof(*rss) * (svc->cnt + 1));  in rs_svc_grow_sets()
3958 		memcpy(contexts, svc->contexts, svc->context_size * (svc->cnt + 1));  in rs_svc_grow_sets()
3961 	free(svc->rss);  in rs_svc_grow_sets()
3962 	svc->rss = rss;  in rs_svc_grow_sets()
3963 	svc->contexts = contexts;  in rs_svc_grow_sets()
3970 static int rs_svc_add_rs(struct rs_svc *svc, struct rsocket *rs)  in rs_svc_add_rs()  argument
3974 	if (svc->cnt >= svc->size - 1) {  in rs_svc_add_rs()
3975 		ret = rs_svc_grow_sets(svc, 4);  in rs_svc_add_rs()
3980 	svc->rss[++svc->cnt] = rs;  in rs_svc_add_rs()
3984 static int rs_svc_index(struct rs_svc *svc, struct rsocket *rs)  in rs_svc_index()  argument
3988 	for (i = 1; i <= svc->cnt; i++) {  in rs_svc_index()
3989 		if (svc->rss[i] == rs)  in rs_svc_index()
3995 static int rs_svc_rm_rs(struct rs_svc *svc, struct rsocket *rs)  in rs_svc_rm_rs()  argument
3999 	if ((i = rs_svc_index(svc, rs)) >= 0) {  in rs_svc_rm_rs()
4000 		svc->rss[i] = svc->rss[svc->cnt];  in rs_svc_rm_rs()
4001 		memcpy(svc->contexts + i * svc->context_size,  in rs_svc_rm_rs()
4002 		       svc->contexts + svc->cnt * svc->context_size,  in rs_svc_rm_rs()
4003 		       svc->context_size);  in rs_svc_rm_rs()
4004 		svc->cnt--;  in rs_svc_rm_rs()
4010 static void udp_svc_process_sock(struct rs_svc *svc)  in udp_svc_process_sock()  argument
4014 	read_all(svc->sock[1], &msg, sizeof msg);  in udp_svc_process_sock()
4017 		msg.status = rs_svc_add_rs(svc, msg.rs);  in udp_svc_process_sock()
4020 			udp_svc_fds = svc->contexts;  in udp_svc_process_sock()
4021 			udp_svc_fds[svc->cnt].fd = msg.rs->udp_sock;  in udp_svc_process_sock()
4022 			udp_svc_fds[svc->cnt].events = POLLIN;  in udp_svc_process_sock()
4023 			udp_svc_fds[svc->cnt].revents = 0;  in udp_svc_process_sock()
4027 		msg.status = rs_svc_rm_rs(svc, msg.rs);  in udp_svc_process_sock()
4038 	write_all(svc->sock[1], &msg, sizeof msg);  in udp_svc_process_sock()
4210 	struct rs_svc *svc = arg;  in udp_svc_run()  local
4214 	ret = rs_svc_grow_sets(svc, 4);  in udp_svc_run()
4217 		write_all(svc->sock[1], &msg, sizeof msg);  in udp_svc_run()
4221 	udp_svc_fds = svc->contexts;  in udp_svc_run()
4222 	udp_svc_fds[0].fd = svc->sock[1];  in udp_svc_run()
4225 		for (i = 0; i <= svc->cnt; i++)  in udp_svc_run()
4228 		poll(udp_svc_fds, svc->cnt + 1, -1);  in udp_svc_run()
4230 			udp_svc_process_sock(svc);  in udp_svc_run()
4232 		for (i = 1; i <= svc->cnt; i++) {  in udp_svc_run()
4234 				udp_svc_process_rs(svc->rss[i]);  in udp_svc_run()
4236 	} while (svc->cnt >= 1);  in udp_svc_run()
4250 static void tcp_svc_process_sock(struct rs_svc *svc)  in tcp_svc_process_sock()  argument
4255 	read_all(svc->sock[1], &msg, sizeof msg);  in tcp_svc_process_sock()
4258 		msg.status = rs_svc_add_rs(svc, msg.rs);  in tcp_svc_process_sock()
4261 			tcp_svc_timeouts = svc->contexts;  in tcp_svc_process_sock()
4262 			tcp_svc_timeouts[svc->cnt] = rs_get_time() +  in tcp_svc_process_sock()
4267 		msg.status = rs_svc_rm_rs(svc, msg.rs);  in tcp_svc_process_sock()
4272 		i = rs_svc_index(svc, msg.rs);  in tcp_svc_process_sock()
4286 	write_all(svc->sock[1], &msg, sizeof msg);  in tcp_svc_process_sock()
4306 	struct rs_svc *svc = arg;  in tcp_svc_run()  local
4312 	ret = rs_svc_grow_sets(svc, 16);  in tcp_svc_run()
4315 		write_all(svc->sock[1], &msg, sizeof msg);  in tcp_svc_run()
4319 	tcp_svc_timeouts = svc->contexts;  in tcp_svc_run()
4320 	fds.fd = svc->sock[1];  in tcp_svc_run()
4326 			tcp_svc_process_sock(svc);  in tcp_svc_run()
4330 		for (i = 1; i <= svc->cnt; i++) {  in tcp_svc_run()
4332 				tcp_svc_send_keepalive(svc->rss[i]);  in tcp_svc_run()
4334 					now + svc->rss[i]->keepalive_time;  in tcp_svc_run()
4340 	} while (svc->cnt >= 1);  in tcp_svc_run()