Add ChiakiStopPipe

lib/CMakeLists.txt

@@ -23,7 +23,8 @@ set(HEADER_FILES
 		include/chiaki/frameprocessor.h
 		include/chiaki/seqnum.h
 		include/chiaki/discovery.h
-		include/chiaki/congestioncontrol.h)
+		include/chiaki/congestioncontrol.h
+		include/chiaki/stoppipe.h)
 
 set(SOURCE_FILES
 		src/common.c
@@ -49,7 +50,8 @@ set(SOURCE_FILES
 		src/videoreceiver.c
 		src/frameprocessor.c
 		src/discovery.c
-		src/congestioncontrol.c)
+		src/congestioncontrol.c
+		src/stoppipe.c)
 
 add_subdirectory(protobuf)
 include_directories("${NANOPB_SOURCE_DIR}")

lib/include/chiaki/stoppipe.h

@@ -0,0 +1,45 @@
+/*
+ * This file is part of Chiaki.
+ *
+ * Chiaki is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Chiaki is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Chiaki.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef CHIAKI_STOPPIPE_H
+#define CHIAKI_STOPPIPE_H
+
+#include "common.h"
+
+#include <stdint.h>
+#include <stddef.h>
+#include <sys/time.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct chiaki_stop_pipe_t
+{
+	int fds[2];
+} ChiakiStopPipe;
+
+CHIAKI_EXPORT ChiakiErrorCode chiaki_stop_pipe_init(ChiakiStopPipe *stop_pipe);
+CHIAKI_EXPORT void chiaki_stop_pipe_fini(ChiakiStopPipe *stop_pipe);
+CHIAKI_EXPORT void chiaki_stop_pipe_stop(ChiakiStopPipe *stop_pipe);
+CHIAKI_EXPORT ChiakiErrorCode chiaki_stop_pipe_select_single(ChiakiStopPipe *stop_pipe, int fd, struct timeval *timeout);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // CHIAKI_STOPPIPE_H

lib/include/chiaki/takion.h

@@ -23,6 +23,7 @@
 #include "log.h"
 #include "gkcrypt.h"
 #include "seqnum.h"
+#include "stoppipe.h"
 
 #include <netinet/in.h>
 #include <stdbool.h>
@@ -98,7 +99,7 @@ typedef struct chiaki_takion_t
 	void *av_cb_user;
 	int sock;
 	ChiakiThread thread;
-	int stop_pipe[2];
+	ChiakiStopPipe stop_pipe;
 	struct timeval recv_timeout;
 	int send_retries;
 	uint32_t tag_local;

lib/src/stoppipe.c

@@ -0,0 +1,75 @@
+/*
+ * This file is part of Chiaki.
+ *
+ * Chiaki is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Chiaki is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Chiaki.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include <chiaki/stoppipe.h>
+
+#include <unistd.h>
+#include <fcntl.h>
+#include <sys/socket.h>
+
+
+CHIAKI_EXPORT ChiakiErrorCode chiaki_stop_pipe_init(ChiakiStopPipe *stop_pipe)
+{
+	int r = pipe(stop_pipe->fds);
+	if(r < 0)
+		return CHIAKI_ERR_UNKNOWN;
+
+	r = fcntl(stop_pipe->fds[0], F_SETFL, O_NONBLOCK);
+	if(r == -1)
+	{
+		close(stop_pipe->fds[0]);
+		close(stop_pipe->fds[1]);
+		return CHIAKI_ERR_UNKNOWN;
+	}
+
+	return CHIAKI_ERR_SUCCESS;
+}
+
+CHIAKI_EXPORT void chiaki_stop_pipe_fini(ChiakiStopPipe *stop_pipe)
+{
+	close(stop_pipe->fds[0]);
+	close(stop_pipe->fds[1]);
+}
+
+CHIAKI_EXPORT void chiaki_stop_pipe_stop(ChiakiStopPipe *stop_pipe)
+{
+	write(stop_pipe->fds[1], "\x00", 1);
+}
+
+CHIAKI_EXPORT ChiakiErrorCode chiaki_stop_pipe_select_single(ChiakiStopPipe *stop_pipe, int fd, struct timeval *timeout)
+{
+	fd_set fds;
+	FD_ZERO(&fds);
+	FD_SET(fd, &fds);
+	FD_SET(stop_pipe->fds[0], &fds);
+
+	int nfds = fd;
+	if(stop_pipe->fds[0] > nfds)
+		nfds = stop_pipe->fds[0];
+	nfds++;
+	int r = select(nfds, &fds, NULL, NULL, timeout);
+	if(r < 0)
+		return CHIAKI_ERR_UNKNOWN;
+
+	if(FD_ISSET(stop_pipe->fds[0], &fds))
+		return CHIAKI_ERR_CANCELED;
+
+	if(FD_ISSET(fd, &fds))
+		return CHIAKI_ERR_SUCCESS;
+
+	return CHIAKI_ERR_TIMEOUT;
+}

lib/src/takion.c

@@ -128,21 +128,13 @@ CHIAKI_EXPORT ChiakiErrorCode chiaki_takion_connect(ChiakiTakion *takion, Chiaki
 
 	CHIAKI_LOGI(takion->log, "Takion connecting\n");
 
-	int r = pipe(takion->stop_pipe);
+	ChiakiErrorCode err = chiaki_stop_pipe_init(&takion->stop_pipe);
-	if(r < 0)
+	if(err != CHIAKI_ERR_SUCCESS)
 	{
-		CHIAKI_LOGE(takion->log, "Takion failed to create pipe\n");
+		CHIAKI_LOGE(takion->log, "Takion failed to create stop pipe\n");
 		return CHIAKI_ERR_UNKNOWN;
 	}
 
-	r = fcntl(takion->stop_pipe[0], F_SETFL, O_NONBLOCK);
-	if(r == -1)
-	{
-		CHIAKI_LOGE(takion->log, "Takion failed to fcntl pipe\n");
-		ret = CHIAKI_ERR_UNKNOWN;
-		goto error_pipe;
-	}
-
 	takion->sock = socket(info->sa->sa_family, SOCK_DGRAM, IPPROTO_UDP);
 	if(takion->sock < 0)
 	{
@@ -151,7 +143,7 @@ CHIAKI_EXPORT ChiakiErrorCode chiaki_takion_connect(ChiakiTakion *takion, Chiaki
 		goto error_pipe;
 	}
 
-	r = connect(takion->sock, info->sa, info->sa_len);
+	int r = connect(takion->sock, info->sa, info->sa_len);
 	if(r < 0)
 	{
 		CHIAKI_LOGE(takion->log, "Takion failed to connect: %s\n", strerror(errno));
@@ -168,7 +160,7 @@ CHIAKI_EXPORT ChiakiErrorCode chiaki_takion_connect(ChiakiTakion *takion, Chiaki
 	init_payload.min = TAKION_LOCAL_MIN;
 	init_payload.max = TAKION_LOCAL_MIN;
 	init_payload.tag1 = takion->tag_local;
-	ChiakiErrorCode err = takion_send_message_init(takion, &init_payload);
+	err = takion_send_message_init(takion, &init_payload);
 	if(err != CHIAKI_ERR_SUCCESS)
 	{
 		CHIAKI_LOGE(takion->log, "Takion failed to send init\n");
@@ -255,16 +247,15 @@ CHIAKI_EXPORT ChiakiErrorCode chiaki_takion_connect(ChiakiTakion *takion, Chiaki
 error_sock:
 	close(takion->sock);
 error_pipe:
-	close(takion->stop_pipe[0]);
+	chiaki_stop_pipe_fini(&takion->stop_pipe);
-	close(takion->stop_pipe[1]);
 	return ret;
 }
 
 CHIAKI_EXPORT void chiaki_takion_close(ChiakiTakion *takion)
 {
-	write(takion->stop_pipe[1], "\x00", 1);
+	chiaki_stop_pipe_stop(&takion->stop_pipe);
 	chiaki_thread_join(&takion->thread, NULL);
-	close(takion->stop_pipe[1]);
+	chiaki_stop_pipe_fini(&takion->stop_pipe);
 }
 
 CHIAKI_EXPORT ChiakiErrorCode chiaki_takion_send_raw(ChiakiTakion *takion, uint8_t *buf, size_t buf_size)
@@ -364,34 +355,21 @@ static void *takion_thread_func(void *user)
 
 beach:
 	close(takion->sock);
-	close(takion->stop_pipe[0]);
 	return NULL;
 }
 
 
 static ChiakiErrorCode takion_recv(ChiakiTakion *takion, uint8_t *buf, size_t *buf_size, struct timeval *timeout)
 {
-	fd_set fds;
+	ChiakiErrorCode err = chiaki_stop_pipe_select_single(&takion->stop_pipe, takion->sock, timeout);
-	FD_ZERO(&fds);
+	if(err == CHIAKI_ERR_TIMEOUT)
-	FD_SET(takion->sock, &fds);
+		return err;
-	FD_SET(takion->stop_pipe[0], &fds);
+	if(err != CHIAKI_ERR_SUCCESS)
-
-	int nfds = takion->sock;
-	if(takion->stop_pipe[0] > nfds)
-		nfds = takion->stop_pipe[0];
-	nfds++;
-	int r = select(nfds, &fds, NULL, NULL, timeout);
-	if(r < 0)
 	{
 		CHIAKI_LOGE(takion->log, "Takion select failed: %s\n", strerror(errno));
-		return CHIAKI_ERR_UNKNOWN;
+		return err;
 	}
 
-	if(FD_ISSET(takion->stop_pipe[0], &fds))
-		return CHIAKI_ERR_CANCELED;
-
-	if(FD_ISSET(takion->sock, &fds))
-	{
 	ssize_t received_sz = recv(takion->sock, buf, *buf_size, 0);
 	if(received_sz <= 0)
 	{
@@ -403,9 +381,6 @@ static ChiakiErrorCode takion_recv(ChiakiTakion *takion, uint8_t *buf, size_t *b
 	}
 	*buf_size = (size_t)received_sz;
 	return CHIAKI_ERR_SUCCESS;
-	}
-
-	return CHIAKI_ERR_TIMEOUT;
 }