ofed/librdmacm/rsocket.c

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