Enabling nagle, and adding Linux only TCP flush

Fixing EINTR handling during socket error handling
Using epoll_pwait2 if possible

Author: ktakashi

ext/socket/CMakeLists.txt

@@ -15,6 +15,11 @@ ENDIF()
 # Windows       - WSAPoll
 IF (${CMAKE_SYSTEM_NAME} STREQUAL "Linux")
   SET(SELECTOR_SOURCE selector-epoll.c)
+  CHECK_FUNCTION_EXISTS(epoll_pwait2 HAVE_EPOLL_PWAIT2)
+  IF (HAVE_EPOLL_PWAIT2)
+    MESSAGE(STATUS "Using epoll_wait2")
+    ADD_DEFINITIONS(-DHAVE_EPOLL_PWAIT2)
+  ENDIF()
 ELSEIF(WIN32)
   SET(SELECTOR_SOURCE selector-win.c)
 ELSE()

ext/socket/sagittarius-socket.c

@@ -1,6 +1,6 @@
 /* sagittarius-socket.c                            -*- mode:c; coding:utf-8; -*-
  *
- *   Copyright (c) 2010-2016  Takashi Kato <ktakashi@ymail.com>
+ *   Copyright (c) 2010-2025  Takashi Kato <ktakashi@ymail.com>
  *
  *   Redistribution and use in source and binary forms, with or without
  *   modification, are permitted provided that the following conditions
@@ -500,13 +500,15 @@ SgObject Sg_CreateSocket(int family, int socktype, int protocol)
   return make_socket_inner(fd);
 }
 
-static void disable_nagle(SOCKET fd)
+static void flush_tcp(SOCKET fd)
 {
 #ifdef TCP_NODELAY
-    const int value = 1;
-    /* we ignore the return value here, since this is merely performance
-       optimisation */
-    setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (const char *)&value, sizeof(int));
+  int value = 1;
+  /* we ignore the return value here, since this is merely performance
+     optimisation */
+  setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (const char *)&value, sizeof(int));
+  value = 0;
+  setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (const char *)&value, sizeof(int));
 #endif
 }
 
@@ -541,8 +543,6 @@ SgObject Sg_SocketConnect(SgSocket *socket, SgAddrinfo* addrinfo,
       goto err;
     }
   }
-
-  disable_nagle(socket->socket);
   socket->type = SG_SOCKET_CLIENT;
   socket->address = SG_SOCKADDR(ai_addr(addrinfo));
   socket->node = addrinfo->node;
@@ -708,29 +708,27 @@ SgObject Sg_SocketGetopt(SgSocket *socket, int level, int name, int rsize)
 #endif
 
 #define handleError(who, socket, r)					\
-  do {									\
-    if ((r) < 0) {							\
-      int e = last_error;						\
-      socket->lastError = e;						\
-      switch (e) {							\
-      case EINTR:							\
-	continue;							\
-      case EPIPE:							\
-	if (flags & MSG_NOSIGNAL) {					\
-	  return 0;							\
-	}								\
-	break;								\
-      case NON_BLOCKING_CASE:						\
-      case ETIMEDOUT:/* Windows, maybe we should raise an error here*/	\
-	/* most probably non-blocking socket */				\
-	return (r);							\
+  if ((r) < 0) {							\
+    int e = last_error;							\
+    socket->lastError = e;						\
+    switch (e) {							\
+    case EINTR:								\
+      continue;								\
+    case EPIPE:								\
+      if (flags & MSG_NOSIGNAL) {					\
+	return 0;							\
       }									\
-      raise_socket_error(SG_INTERN(who),				\
-			 Sg_GetLastErrorMessageWithErrorCode(e),	\
-			 Sg_MakeConditionSocket(socket),		\
-			 SG_LIST1(SG_MAKE_INT(e)));			\
+      break;								\
+    case NON_BLOCKING_CASE:						\
+    case ETIMEDOUT:/* Windows, maybe we should raise an error here*/	\
+      /* most probably non-blocking socket */				\
+      return (r);							\
     }									\
-  } while (0)
+    raise_socket_error(SG_INTERN(who),					\
+		       Sg_GetLastErrorMessageWithErrorCode(e),		\
+		       Sg_MakeConditionSocket(socket),			\
+		       SG_LIST1(SG_MAKE_INT(e)));			\
+  }									\
 
 long Sg_SocketReceive(SgSocket *socket, uint8_t *data, long size, int flags)
 {
@@ -776,6 +774,7 @@ long Sg_SocketSend(SgSocket *socket, uint8_t *data, long size, int flags)
     data += ret;
     size -= ret;
   }
+  flush_tcp(socket->socket);
   return sizeSent;
 }
 
@@ -870,7 +869,6 @@ SgObject Sg_SocketAccept(SgSocket *socket)
       break;
     }
   }
-  disable_nagle(fd);
   return make_socket(fd, SG_SOCKET_SERVER, 
 		     make_sockaddr(addrlen, (struct sockaddr *)&addr, TRUE));
 }

ext/socket/sagittarius/socket.scm

@@ -230,7 +230,6 @@
 	(cond ((and (zero? ms) (zero? ns)) "") ;; no timeout
 	      ((zero? ms) (format ", timeout: ~a" t)) ;; less then millis
 	      (else (format ", timeout: ~ams" ms)))))
-	
     (let ((to (socket-get-read-timeout sock)))
       (raise (condition (make-socket-read-timeout-error sock to)
 			(make-who-condition who)

ext/socket/selector-epoll.c

@@ -76,14 +76,17 @@ static SgObject wait_selector(unix_context_t *ctx, int nsock,
 			      int *err)
 {
   int n = nsock + 1, i, c;
-  long millis = -1;
   SgObject r = SG_NIL, cp;
   struct epoll_event *evm, ev;
 
+#ifndef HAVE_EPOLL_PWAIT2
+  long millis = -1;
   if (sp) {
     millis = sp->tv_sec * 1000;
     millis += sp->tv_nsec / 1000000;
   }
+#endif
+
   SG_FOR_EACH(cp, sockets) {
     add_socket_ctx(ctx, SG_CAR(cp));
   }
@@ -92,7 +95,11 @@ static SgObject wait_selector(unix_context_t *ctx, int nsock,
   epoll_ctl(ctx->fd, EPOLL_CTL_ADD, ctx->stop_fd, &ev);
 
   evm = SG_NEW_ATOMIC2(struct epoll_event *, n * sizeof(struct epoll_event));
+#ifndef HAVE_EPOLL_PWAIT2
   c = epoll_wait(ctx->fd, evm, n, millis);
+#else
+  c = epoll_pwait2(ctx->fd, evm, n, sp, NULL);
+#endif
   /*
     EINTR  The call was interrupted by a signal handler before either
     (1) any of the requested events occurred or (2) the
@@ -108,9 +115,8 @@ static SgObject wait_selector(unix_context_t *ctx, int nsock,
   for (i = 0; i < c; i++) {
     if (SG_FALSEP(evm[i].data.ptr)) {
       interrupted_unix_stop(ctx);
-    } else if (SG_PAIRP(evm[i].data.ptr) && evm[i].events == EPOLLIN) {
+    } else if (evm[i].events & EPOLLIN) {
       SgObject slot = SG_OBJ(evm[i].data.ptr);
-      /* SgSocket *sock = SG_SOCKET(SG_CAR(slot)); */
       r = Sg_Cons(slot, r);
     }
   }