diff --git a/examples/fork-test/srt_server+client.c b/examples/fork-test/srt_server+client.c new file mode 100644 index 000000000..d23c8ab63 --- /dev/null +++ b/examples/fork-test/srt_server+client.c @@ -0,0 +1,195 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#define PORT 9000 + +void *run(void *arg) { + char *command = (char*)arg; + pid_t pid = fork(); + if (pid < 0) { + perror("fork (intermediate)"); + exit(EXIT_FAILURE); + } + if (pid > 0) { + // Parent process + printf("[GRANDPARENT %d] waiting for grand-child process pid=%d to finish...\n", + getpid(), pid); + waitpid(pid, NULL, 0); // Wait for intermediate child + printf("[GRANDPARENT] returning\n"); + return 0; + } + // Intermediate process + //srt_cleanup(); + if (setsid() < 0) { + perror("setsid"); + exit(EXIT_FAILURE); + } + pid_t grandchild_pid = fork(); + if (grandchild_pid < 0) { + perror("fork (grandchild)"); + exit(EXIT_FAILURE); + } + if (grandchild_pid > 0) { + printf("[PARENT %d] waiting for 2s with child process pid=%d ...\n", + getpid(), grandchild_pid); + // Intermediate process exits immediately + sleep(2); + printf("[PARENT] exiting\n"); + exit(0); + } + // Grandchild process + // Redirect stdin to /dev/null + printf("[CHILD %d] Preparing descriptors...\n", getpid()); + int devnull = open("/dev/null", O_RDONLY); + if (devnull >= 0) { + dup2(devnull, STDIN_FILENO); + close(devnull); + } else { + perror("open /dev/null"); + } + // Redirect stdout to stderr + dup2(STDERR_FILENO, STDOUT_FILENO); + // Execute the command + printf("[CHILD] Executing process '%s'...\n", command); + execl("/bin/sh", "sh", "-c", command, (char *)NULL); + // If execl fails + perror("execl"); + exit(EXIT_FAILURE); + return 0; +} + +void *server_thread_f(void *arg) { + SRTSOCKET serv_sock = srt_create_socket(); + if (serv_sock == SRT_INVALID_SOCK) { + fprintf(stderr, "Error creating SRT socket: %s\n", srt_getlasterror_str()); + return 0; + } + struct sockaddr_in sa; + memset(&sa, 0, sizeof sa); + sa.sin_family = AF_INET; + sa.sin_port = htons(PORT); + sa.sin_addr.s_addr = INADDR_ANY; + if (srt_bind(serv_sock, (struct sockaddr*)&sa, sizeof sa) == SRT_ERROR) { + fprintf(stderr, "Error: srt_bind: %s\n", srt_getlasterror_str()); + return 0; + } + if (srt_listen(serv_sock, 5) == SRT_ERROR) { + fprintf(stderr, "Error: srt_listen: %s\n", srt_getlasterror_str()); + return 0; + } + printf("SRT server is listening on port %d...\n", PORT); + struct sockaddr_in client_addr; + int addr_len = sizeof(client_addr); + SRTSOCKET client_sock = srt_accept(serv_sock, (struct sockaddr*)&client_addr, &addr_len); + if (client_sock == SRT_INVALID_SOCK) { + fprintf(stderr, "Error: srt_accept: %s\n", srt_getlasterror_str()); + return 0; + } + printf("Client connected via SRT !\n"); + char buffer[1500]; + int bytes = srt_recv(client_sock, buffer, sizeof(buffer)); + if (bytes > 0) { + buffer[bytes] = '\0'; + printf("Message received: %s\n", buffer); + const char resp [] = "We read you!"; + srt_send(client_sock, resp, (sizeof resp)-1); + } else { + printf("Error: reading from srt_recv: %s.\n", srt_getlasterror_str()); + } + // run("date > /tmp/res"); + printf("Server: sleep(1)\n"); + sleep(1); + // Nettoyage + printf("Server: closing SRT sockets\n"); + srt_close(client_sock); + srt_close(serv_sock); + return 0; +} + +#define SERVER_IP "127.0.0.1" +#define SERVER_PORT 9000 +void *client_thread_f(void *arg) { + SRTSOCKET client_sock = srt_create_socket(); + if (client_sock == SRT_INVALID_SOCK) { + fprintf(stderr, "Error creating a socket: %s\n", srt_getlasterror_str()); + return 0; + } + struct sockaddr_in sa; + memset(&sa, 0, sizeof sa); + sa.sin_family = AF_INET; + sa.sin_port = htons(SERVER_PORT); + inet_pton(AF_INET, SERVER_IP, &sa.sin_addr); + if (srt_connect(client_sock, (struct sockaddr*)&sa, sizeof(sa)) == SRT_ERROR) { + fprintf(stderr, "Error: srt_connect: %s\n", srt_getlasterror_str()); + return 0; + } + printf("Connected to SRT server %s:%d\n", SERVER_IP, SERVER_PORT); + const char* message = "Hello from SRT client!"; + int bytes = srt_send(client_sock, message, strlen(message)); + if (bytes == SRT_ERROR) { + fprintf(stderr, "Sending error: %s\n", srt_getlasterror_str()); + } else { + printf("Message sent: %s\n", message); + } + + while (1) + { + char buffer[1500]; + int nb = srt_recv(client_sock, buffer, sizeof(buffer)); + if (nb <= 0) + { + printf("Closed from the server !\n"); + srt_close(client_sock); + break; + } + buffer[nb] = 0; + printf("Server has sent: %s\n", buffer); + } + return 0; +} + +int main() { + if (srt_startup() != 0) { + fprintf(stderr, "Error initializing SRT.\n"); + return 1; + } + + pthread_t server_thread; + if (pthread_create(&server_thread, NULL, server_thread_f, NULL) != 0) { + perror("failed to create server thread"); + exit(EXIT_FAILURE); + } + + pthread_t client_thread; + if (pthread_create(&client_thread, NULL, client_thread_f, NULL) != 0) { + perror("failed to create client thread"); + exit(EXIT_FAILURE); + } + + const int run_threads_count = 10; + pthread_t run_threads[run_threads_count]; + for (int i = 0; i < run_threads_count; ++i) { + if (pthread_create(&run_threads[i], NULL, run, "date > /tmp/res") != 0) { + perror("failed to create run thread"); + exit(EXIT_FAILURE); + } + } + for (int i = 0; i < run_threads_count; ++i) { + if (pthread_join(run_threads[i], NULL) != 0) { + perror("failed joining run thread"); + } + } + + pthread_join(server_thread, NULL); + pthread_join(client_thread, NULL); + + printf("Server: cleanup\n"); + srt_cleanup(); + printf("Server: exit\n"); + return 0; +} diff --git a/srtcore/core.cpp b/srtcore/core.cpp index e4577a27d..ded5d7d66 100644 --- a/srtcore/core.cpp +++ b/srtcore/core.cpp @@ -6461,17 +6461,11 @@ void srt::CUDT::addressAndSend(CPacket& w_pkt) m_pSndQueue->sendto(m_PeerAddr, w_pkt, m_SourceAddr); } -// [[using maybe_locked(m_GlobControlLock, if called from breakSocket_LOCKED, usually from GC)]] -// [[using maybe_locked(m_parent->m_ControlLock, if called from srt_close())]] -bool srt::CUDT::closeInternal() ATR_NOEXCEPT +// This is specifically for doing only the most important +// part that is independent on the current state, which is +// the only part that can be cleaned up after fork(). +bool srt::CUDT::closeBasic() ATR_NOEXCEPT { - // NOTE: this function is called from within the garbage collector thread. - - if (!m_bOpened) - { - return false; - } - // IMPORTANT: // This function may block indefinitely, if called for a socket // that has m_bBroken == false or m_bConnected == true. @@ -6573,6 +6567,23 @@ bool srt::CUDT::closeInternal() ATR_NOEXCEPT // Inform the threads handler to stop. m_bClosing = true; + return true; +} + +// [[using maybe_locked(m_GlobControlLock, if called from breakSocket_LOCKED, usually from GC)]] +// [[using maybe_locked(m_parent->m_ControlLock, if called from srt_close())]] +bool srt::CUDT::closeInternal() ATR_NOEXCEPT +{ + // NOTE: this function is called from within the garbage collector thread. + + if (!m_bOpened) + { + return false; + } + + if (!closeBasic()) + return false; + HLOGC(smlog.Debug, log << CONID() << "CLOSING STATE (closing=true). Acquiring connection lock"); ScopedLock connectguard(m_ConnectionLock); @@ -6646,7 +6657,7 @@ bool srt::CUDT::closeInternal() ATR_NOEXCEPT bool srt::CUDT::closeAtFork() ATR_NOEXCEPT { m_bShutdown = true; - return closeInternal(); + return closeBasic(); } int srt::CUDT::receiveBuffer(char *data, int len) diff --git a/srtcore/core.h b/srtcore/core.h index a4a9acaee..1beb55342 100644 --- a/srtcore/core.h +++ b/srtcore/core.h @@ -748,6 +748,7 @@ class CUDT bool closeInternal() ATR_NOEXCEPT; bool closeAtFork() ATR_NOEXCEPT; + bool closeBasic() ATR_NOEXCEPT; void updateBrokenConnection(); void completeBrokenConnectionDependencies(int errorcode);