epoll demo
lazarus / 38-Networking and Web Programming

----------------------------------------------------------------------
x-dsl · 2010-05-06 22:02

hei,every body,i worte a demo with epoll.
do you wand it?

----------------------------------------------------------------------
x-dsl · 2010-05-08 01:15

program epoll_server;

{$mode objfpc}{$H+}

uses
  {$IFDEF UNIX}{$IFDEF UseCThreads}
  cthreads,
  {$ENDIF}{$ENDIF}
  Classes, SysUtils, CustApp, Linux, LibC, IniFiles
  { you can add units after this };

type

  { TEPoll_Server }

  TEPoll_Server = class(TCustomApplication)
  protected
    hlBuf_Send :THashedStringList;
    procedure DoRun; override;
    function setnonblocking(sockfd :Integer) :Integer;
    function handle_message(new_fd :Integer; EventID :Integer)
:Integer;
  public
    constructor Create(TheOwner: TComponent); override;
    destructor Destroy; override;
  end;

{ TEPoll_Server }
const
  MAXBUF       = 2;
  MAXEPOLLSIZE = 10000;
  EPOLLONESHOT = $40000000; { Sets the One-Shot behaviour for the
associated file descriptor. }

function TEPoll_Server.setnonblocking(sockfd :Integer) :Integer;
var
  iResult :Integer;
begin
  iResult := fcntl(sockfd, F_SETFL, O_NONBLOCK);
  Result  := iResult;
end;


//handle_message - 处理每个 socket 上的消息收发
function TEPoll_Server.handle_message(new_fd :Integer; EventID
:Integer) :Integer;
var
  buf  :Array[0..MAXBUF-1] of Char;
  len  :Integer;
  iInx :Integer;
  sTmp :String;
begin
    //* 开始处理每个新连接上的数据收发 */
    //memset(@buf, 0, MAXBUF);

    //读到的数据有可能大于申请的数据，这时候要反复读
    //iInx := 1;
  if EventID and EPOLLIN <> 0 then

    begin
    SetLength(sTMp,0);
    repeat
      len := LibC.Recv(new_fd,buf[0],MAXBUF,0);
      if len > 0 then
      begin
        iInx := Length(sTmp) + 1;
        SetLength(sTMp,Length(sTMp) + len);
        Move(buf[0],sTmp[iInx],len);
      end;
      Writeln('信息 : 句柄[' + IntToStr(new_fd) + '],本次读取到的数据长度[' +
IntTostr(len) + ']');

    until
      (len <= 0) or (len < MAXBUF);


    if (Length(sTMp) > 0) then
    begin
      //SetLength(sTMp,len);

      Writeln('消息 : 句柄[' + IntToStr(new_fd) + '],长度[' +
IntToStr(Length(sTMp)) + ']>>>' + Trim(sTmp));
      hlBuf_Send.Add(IntToStr(new_fd) + '=' + sTMp);
      //if libc.send(new_fd,sTmp[1],Length(sTMp),0) <> Length(sTMp)
then
      //  Writeln('错误 : 消息发送到[' + IntToStr(new_fd) + ']失败,' +
IntToStr(ErrNO) + ',' + StrError(ErrNO));
    end
    else if (len = SOCKET_ERROR) and (ErrNO <> EAGAIN) then
    begin
      ShowException(Exception.Create('错误 : 消息接收失败:' + IntToStr(ErrNO)
+ ',' + StrError(ErrNO)));
      __close(new_fd);
      Result := -1
    end
    else if (len = 0) then
    begin
      Writeln('消息 : 句柄,' + IntToStr(new_fd) + '已断开与服务器的链接.');
      __close(new_fd);
      Result := -1
    end;
    //* 处理每个新连接上的数据收发结束 */
    Result := len;

    end;

  if EventID and EPOLLOUT <> 0 then
  begin
    //iInx := hlBuf_Send.IndexOfName();
    iInx := hlBuf_Send.IndexOfName(IntToStr(new_fd));
    if iInx > -1 then
    begin
      sTMp := hlBuf_Send.ValueFromIndex[iInx];
      libc.send(new_fd,sTmp[1],Length(sTMp),0);
      hlBuf_Send.Delete(iInx);
    end;
  end;

end;

procedure TEPoll_Server.DoRun;
var
  ErrorMsg: String;
  my_addr, their_addr :sockaddr_in;
  listener, new_fd, kdpfd, nfds, n, ret, curfds :Integer;
  len     :Integer;
  myport  :Integer;
  ev      :epoll_event;
  events  :array [0..MAXEPOLLSIZE-1] of epoll_event;
  rt      :rlimit;
  sTmp    :String;
  opt      :Integer;
begin
  hlBuf_Send := THashedStringList.Create;

  // quick check parameters
  ErrorMsg:=CheckOptions('p','port');
  if ErrorMsg<>'' then begin
    ShowException(Exception.Create(ErrorMsg));
    Terminate;
    Exit;
  end;

  //获取参数  -p 端口号
  sTmp   := Trim(GetOptionValue('p'));
  myport := -1;
  myport := StrToIntDef(sTmp, -1);
  if (myPort < $0001) or (myPort > $FFFF) then
  begin
    ShowException(Exception.Create('错误 : 无效的端口号:' + sTmp));
    Terminate;
    Exit;
  end;


  //设置每个进程允许打开的最大文件数
  rt.rlim_max := MAXEPOLLSIZE;
  rt.rlim_cur := MAXEPOLLSIZE;
  if (setrlimit(RLIMIT_NOFILE, @rt) = -1) then
  begin
    ShowException(Exception.Create('错误 : 设置系统资源参数失败:' +
IntToStr(ErrNO) + ',' + StrError(ErrNO)));
    Terminate;
    Exit;
  end
  else
    Writeln('信息 : 设置系统资源参数成功！');

  //创建文件句柄
  listener := socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
  if listener =  SOCKET_ERROR then
  begin
    ShowException(Exception.Create('错误 : 创建Socket句柄失败:' +
IntToStr(ErrNO) + ',' + StrError(ErrNO)));
    Terminate;
    Exit;
  end;

  //设置非堵塞模式
  if setnonblocking(listener) =  SOCKET_ERROR then
  begin
    ShowException(Exception.Create('错误 : 设置非堵塞模式失败:' + IntToStr(ErrNO)
+ ',' + StrError(ErrNO)));
    Terminate;
    Exit;
  end;

  //回收操作系统曾经打开过的端口资源
  opt := 1;
  if setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, @OPt, sizeof
(opt))  =  SOCKET_ERROR then
  begin
    ShowException(Exception.Create('错误 : 缓冲区清除失败:' + IntToStr(ErrNO) +
',' + StrError(ErrNO)));
    Terminate;
    Exit;
  end;

  //绑定地址与端口
  memset(@my_addr, 0, SizeOf(my_addr));
  my_addr.sa_family := AF_INET;
  my_addr.sin_port  := htons(myPort);
  if LibC.bind(listener, my_addr, SizeOf(my_addr)) = SOCKET_ERROR then
  begin
    ShowException(Exception.Create('错误 : 绑定地址与端口失败:' + IntToStr(ErrNO)
+ ',' + StrError(ErrNO)));
    Terminate;
    Exit;
  end;

  // 监听失败
  if LibC.listen(listener, 128) = SOCKET_ERROR then
  begin
    ShowException(Exception.Create('错误 : 监听失败:' + IntToStr(ErrNO) +
',' + StrError(ErrNO)));
    Terminate;
    Exit;
  end;



  //创建EPOLL
  kdpfd      := epoll_create(MAXEPOLLSIZE);
  len        := SizeOf(sockaddr_in);
  ev.events  := EPOLLIN or EPOLLOUT or EPOLLET;
  //ev.events  := EPOLLIN or EPOLLOUT;
  ev.data.fd := listener;
  if epoll_ctl(kdpfd, EPOLL_CTL_ADD, listener, @ev) = SOCKET_ERROR
then
  begin
    ShowException(Exception.Create('错误 : 创建EPOLL失败:' + IntToStr(ErrNO)
+ ',' + StrError(ErrNO)));
    Terminate;
    Exit;
  end;
  curfds     := 1;

  writeln('信息 : 服务已启动，端口 : ' + IntToStr(myPort));

  while true do
  begin
    //等待有事件发生
    nfds := epoll_wait(kdpfd, events, curfds, -1);
    if (nfds = -1) then
    begin
      ShowException(Exception.Create('错误 : epoll_wait失败:' +
IntToStr(ErrNO) + ',' + StrError(ErrNO)));
      //break;
    end;

    //处理所有事件 */
    for n := 0 to nfds -1 do
    begin
      if (events[n].data.fd = listener) then
      begin
        new_fd := accept(listener, @their_addr, @len);
        if (new_fd < 0) then
        begin
          ShowException(Exception.Create('错误 : 接入传入连接时发生:' +
IntToStr(ErrNO) + ',' + StrError(ErrNO)));
          continue;
        end
        else
        begin
          Writeln('信息 : [' + inet_ntoa(their_addr.sin_addr) + ':' +
IntToStr(ntohs(their_addr.sin_port)) + ']连接到服务器,句柄:' +
IntToStr(new_fd));
          setnonblocking(new_fd);
          ev.events  := EPOLLIN or EPOLLOUT or EPOLLET;
          //ev.events  := EPOLLIN or EPOLLOUT;
          ev.data.fd := new_fd;
          if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, new_fd, @ev) =
SOCKET_ERROR) then
          begin
            Writeln('错误 : 把Socket' + IntToStr(new_fd) + '加入 epoll
失败！,' + IntToStr(ErrNO) + StrError(ErrNO));
            __close(new_fd);
          end;


        end;

      end
      else
      begin
        Writeln('信息 : Epoll事件-->' + IntToStr(events[n].Events));
        ret := handle_message(events[n].data.fd,events[n].Events);
        //座席断开
        if (ret = SOCKET_ERROR) and (ErrNO <> EAGAIN) then  // (errno
<> 11)
        begin
          epoll_ctl(kdpfd, EPOLL_CTL_DEL, events[n].data.fd,@ev);
          //curfds := curfds - 1;

        end;

      end;
    end;


  end;
  __close(listener);
  Terminate;
end;

constructor TEPoll_Server.Create(TheOwner: TComponent);
begin
  inherited Create(TheOwner);
  StopOnException:=True;
end;

destructor TEPoll_Server.Destroy;
begin
  inherited Destroy;
end;



var
  Application: TEPoll_Server;

{$IFDEF WINDOWS}{$R epoll_server.rc}{$ENDIF}

begin
  Application:=TEPoll_Server.Create(nil);
  Application.Title:='TEPoll_Server';
  Application.Run;
  Application.Free;
end.

----------------------------------------------------------------------
Almindor · 2010-05-08 18:44

Nice  Epoll is already  utilized in lNet

----------------------------------------------------------------------
captian jaster · 2010-05-08 23:30

what is this? can someone explain?

----------------------------------------------------------------------
Almindor · 2010-05-09 09:28

Epoll is a linux event watching mechanism (specific to linux) which is
more effective than the old select() or poll().

lNet (the networking library) uses epoll on linux platforms if
available (and other effective calls on freeBSD for example it uses
kqueue)

----------------------------------------------------------------------
marcov · 2010-05-10 16:58

Note that as long as it uses unit libc it only will work on linux/x86.
Try to eliminate that dependancy to also give it a chance to work with
Linux on other architectures.

----------------------------------------------------------------------
jixian.yang · 2011-01-27 15:48

Thanks a lot.

Maybe Synapse need it to improve its performance.

So we can use epoll on linux server and use IOCP on windows server.

----------------------------------------------------------------------
Tz · 2019-07-24 16:17

Does anybody knows how to handle accept timeout?

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i program
epollserver; // http://man7.org/linux/man-
pages/man7/epoll.7.html{$mode objfpc}{$h+}uses CMem, CThreads,
BaseUnix, Termio, Linux, Sockets; const //  linux.pp
http://man7.org/linux/man-pages/man2/epoll_ctl.2.html//  EPOLLIN     
  = $00000001; { The associated file is available for read(2)
operations. }//  EPOLLPRI        = $00000002; { There is urgent data
available for read(2) operations. }//  EPOLLOUT        = $00000004; {
The associated file is available for write(2) operations. }//
 EPOLLERR        = $00000008; { Error condition happened on the
associated file descriptor. }//  EPOLLHUP        = $00000010; { Hang
up happened on the associated file descriptor. }//  need to update
linux.pp from epoll.h    EPOLLRDNORM     = $00000040;    EPOLLRDBAND 
  = $00000080;    EPOLLWRNORM     = $00000100;    EPOLLWRBAND     =
$00000200;    EPOLLMSG        = $00000400;    EPOLLRDHUP      =
$00002000; { (since Linux 2.6.17)                                 
 Stream socket peer closed connection, or shut down writing          
                       half of connection. }    EPOLLEXCLUSIVE  =
$10000000; { 1u << 28 note: thundering herd                          
       (since Linux 4.5)                                   Sets an
exclusive wakeup mode for the epoll file descriptor                  
               that is being attached to the target file descriptor,
fd. }    EPOLLWAKEUP     = $20000000; { 1u << 29 (since Linux 3.5)   
                              If EPOLLONESHOT and EPOLLET are clear
and the process has the                                 
 CAP_BLOCK_SUSPEND capability, ensure that the system does not       
                          enter "suspend" or "hibernate" while this
event is pending or                                   being processed.
}//  EPOLLONESHOT    = $40000000; { Sets the One-Shot behaviour for
the associated file descriptor. }//  EPOLLET         = $80000000; {
Sets  the  Edge  Triggered  behaviour  for  the  associated file
descriptor. } type   TEPollEvents = array of EPoll_Event;   TEPoll =
class (TObject)  private    FEpfd       :THandle;    FEvents    
:TEPollEvents;    function    Add   (const AFd :THandle; AEvents
:UInt32) :TError;    function    Modify(const AFd :THandle; AEvents
:UInt32) :TError;    function    Delete(const AFd :THandle; AEvents
:UInt32) :TError;  public    const      MAX_EVENTS = 64;   
constructor Create;    destructor  Destroy; override;    function  
 Listen(const AFd :THandle) :TError;    function    Accept(const AFd
:THandle) :TError;    function    Close (const AFd :THandle) :TError; 
  function    Wait  (const ATimeout :Int32) :TError;    function  
 IsType(const AEvent :EPoll_Event; ATypes :UInt32) :Boolean;   
procedure   Check (const AEvent :EPoll_Event);    property    Events
:TEPollEvents read FEvents;  end;   TServer = class (TObject) 
private    FEPoll      :TEPoll;    FServer     :TSocket;    FPort    
 :UInt32;    FError      :TError;  public    constructor Create(const
APort :UInt32);    destructor  Destroy; override;    function  
 Disconnect(const ASocket :TSocket) :TError;    function    Receive  
(const ASocket :TSocket; var   ABuffer :String) :TError;    function 
 Send      (const ASocket :TSocket; const ABuffer :String) :TError;  
procedure   Listen;    property    Error :TError read FError; 
end; constructor TEPoll.Create;  begin    SetLength(FEvents,
MAX_EVENTS);    FEpfd := EPoll_create(MAX_EVENTS);  end; destructor
 TEPoll.Destroy;  begin    fpClose(FEpfd);    inherited Destroy; 
end; function    TEPoll.Add(const AFd :THandle; AEvents :UInt32)
:TError;  var event :EPoll_Event;  begin    event.data.fd := AFd;   
event.events  := AEvents;    result := EPoll_ctl(FEpfd, EPoll_CTL_ADD,
event.data.fd, @event);  end; function    TEPoll.Modify(const AFd
:THandle; AEvents :UInt32) :TError;  var event :EPoll_Event;  begin  
event.data.fd := AFd;    event.events  := AEvents;    result :=
EPoll_ctl(FEpfd, EPoll_CTL_MOD, event.data.fd, @event);  end; function
   TEPoll.Delete(const AFd :THandle; AEvents :UInt32) :TError;  var
event :EPoll_Event;  begin    event.data.fd := AFd;    event.events
 := AEvents;    result := EPoll_ctl(FEpfd, EPoll_CTL_DEL,
event.data.fd, @event);  end; function    TEPoll.Listen(const AFd
:THandle) :TError;  begin    // blocking socket    result :=
self.Add(AFd, EPOLLIN); // Level  end; function    TEPoll.Accept(const
AFd :THandle) :TError;  var flags :Int32;  begin    flags := 1;   
fpIOCtl(AFd, FIONBIO, @flags); // non blocking termio    result :=
self.Add(AFd, EPOLLIN + EPOLLET + EPOLLERR + EPOLLHUP + EPOLLRDHUP);
// Edge  end; function    TEPoll.Close(const AFd :THandle) :TError; 
begin    result := self.Delete(AFd, EPOLLIN); // dont now how 
end; function    TEPoll.Wait(const ATimeout: Int32): TError;  begin  
result := EPoll_wait(FEpfd, @FEvents[0], MAX_EVENTS, ATimeout); 
end; function    TEPoll.IsType(const AEvent :EPoll_Event; ATypes
:UInt32) :Boolean;  begin    result := (AEvent.events and ATypes) =
ATypes;  end; procedure   TEPoll.Check(const AEvent :EPoll_Event); 
begin    Write('fd: ', AEvent.data.fd, ' events: ', AEvent.events);  
if IsType(AEvent, EPOLLIN       ) then Write(' EPOLLIN');    if
IsType(AEvent, EPOLLPRI      ) then Write(' EPOLLPRI');    if
IsType(AEvent, EPOLLOUT      ) then Write(' EPOLLOUT');    if
IsType(AEvent, EPOLLERR      ) then Write(' EPOLLERR');    if
IsType(AEvent, EPOLLHUP      ) then Write(' EPOLLHUP');    if
IsType(AEvent, EPOLLRDNORM   ) then Write(' EPOLLRDNORM');    if
IsType(AEvent, EPOLLRDBAND   ) then Write(' EPOLLRDBAND');    if
IsType(AEvent, EPOLLWRNORM   ) then Write(' EPOLLWRNORM');    if
IsType(AEvent, EPOLLWRBAND   ) then Write(' EPOLLWRBAND');    if
IsType(AEvent, EPOLLMSG      ) then Write(' EPOLLMSG');    if
IsType(AEvent, EPOLLRDHUP    ) then Write(' EPOLLRDHUP');    if
IsType(AEvent, EPOLLEXCLUSIVE) then Write(' EPOLLEXCLUSIVE');    if
IsType(AEvent, EPOLLWAKEUP   ) then Write(' EPOLLWAKEUP');    if
IsType(AEvent, EPOLLONESHOT  ) then Write(' EPOLLONESHOT');    if
IsType(AEvent, EPOLLET       ) then Write(' EPOLLET');    WriteLn; 
end; constructor TServer.Create(const APort :UInt32);  begin    FPort
 := APort;    FEPoll := TEPoll.Create;  end; destructor
 TServer.Destroy;  begin    FEPoll.Free;  end; function  
 TServer.Disconnect(const ASocket :TSocket) :TError;  begin   
WriteLn('disconnect ', ASocket);    FEPoll.Close(ASocket);   
fpShutdown(ASocket, SHUT_WR);    fpClose(ASocket);    result :=
SocketError;  end; function    TServer.Receive(const ASocket :TSocket;
var ABuffer :String) :TError;  var readlen :Int32;  begin    if
fpIOCtl(ASocket, FIONREAD, @readlen) <> 0 then      ABuffer := ''   
else      begin        SetLength(ABuffer, readlen);       
fpRecv(ASocket, PChar(ABuffer), readlen, 0);        WriteLn('receive:
', ABuffer);      end;    result := SocketError;  end; function  
 TServer.Send(const ASocket :TSocket; const ABuffer :String) :TError; 
begin    WriteLn('send: ', ABuffer);    fpSend(ASocket,
PChar(ABuffer), Length(ABuffer), 4);    result := SocketError; 
end; procedure   TServer.Listen;  var    i, n,    addrSize :Int32;   
event    :EPoll_Event;    client   :TSocket;    addr    
:TInetSockAddr;    msg      :String;  begin     FServer :=
fpSocket(AF_INET, SOCK_STREAM, 0);     if FServer = -1 then     
begin        FError := SocketError;        Exit;      end;    
addr.sin_family       := AF_INET;    addr.sin_port         :=
NToHS(FPort);    addr.sin_addr.s_addr  := 0;    addrSize            
 := SizeOf(addr);     if fpBind(FServer, @addr, addrSize) <> 0 then  
  begin        FError := SocketError;        Exit;      end;     if
fpListen(FServer, 128) <> 0 then      begin        FError :=
SocketError;        Exit;      end;     if FEPoll.Listen(FServer) = -1
then      begin        FError := fpGetErrNo;        Exit;      end;  
 FError := 0;    repeat      n := FEPoll.Wait(1000);      for i:=0 to
n-1 do        begin          event := FEPoll.Events[i];         
FEPoll.Check(event);           if   FEPoll.IsType(event, EPOLLERR)   
       or FEPoll.IsType(event, EPOLLHUP)            or
FEPoll.IsType(event, EPOLLRDHUP) then             
self.Disconnect(event.data.fd)          else            begin        
    if event.data.fd = FServer then                begin             
   client := fpAccept(FServer, @addr, @addrSize);                  if
client <> -1 then                    FEPoll.Accept(client);  // how to
handle accept timeout ?                end              else         
     begin                  if FEPoll.IsType(event, EPOLLIN) then    
              begin                      self.Receive(event.data.fd,
msg);                      self.Send(event.data.fd, 'echo: ' + msg); 
                   if Length(msg) < 5 then                       
self.Disconnect(event.data.fd);                    end;              
end;            end;        end;    until FError <> 0;    
self.Disconnect(FServer);  end; var server :TServer; begin  server :=
TServer.Create(8000);  server.Listen;  server.Free;end.

----------------------------------------------------------------------
zamronypj · 2019-08-01 01:14

@Tz

I am also still learning on epoll API so CMIIW.

What I learned so far during implementing my epoll-based socket server

https://github.com/fanoframework/fano/blob/master/src/Libs/Socket/Epol
lSocketSvrImpl.pas

epoll_wait() will block until one or more file descriptors in epoll
interest list is signaled (returns > 0) or timeout (returns =0) or
something is wrong (returns = -1). From your code, you add listen
socket to epoll interest list with EPOLLIN, which means, by the time
epoll_wait() returns > 0, you can assume that fpAccept(FServer, ...)
call (line 260) will not block, so timeout is irrelevant.

Also from your code, after accepting client connection, you call
FEpoll.Accept(), see line 262, to  add client socket file descriptors
to epoll interest list using epoll_ctl() will not block, so again
timeout is relevant here.

P.S., actually TEpoll.Accept() name is misleading as it is not really
accepting client connection, merely add client socket to epoll
interest list.

----------------------------------------------------------------------
Tz · 2019-08-01 18:40

for x-dsl thank you for starting this forum, its been long time and so
many changes.

@zamronypj you are correct its missleading, thank you.


epoll for me its a bit confusing, taken from linux.pp and help from
rtl.chm


  EPoll_Data = record
    case Integer of
      0: (ptr   :pointer; );
      1: (fd    :cint;    );
      2: (u32   :cuint;   );
      3: (u64   :cuint64; );
  end;

  EPoll_Event = packed record
    Events      :cuint32;         // should be event types
    Data        :EPoll_Data;
  end;

  function epoll_create(
    size        :cint
  ) :cint;

  function epoll_ctl(
    epfd        :cint;
    op          :cint;
    fd          :cint;
    event       :PEPoll_Event
  ) :cint;

  function epoll_wait(
    epfd        :cint;
    events      :PEpoll_Event;    // better use event list
    maxevents   :cint;
    timeout     :cint
  ) :cint;



first thing is events in EPoll_Event record and events in epoll_wait
function
also as if it recursive in event there is also event

second is fd, it can be socket, while in linux everything is file,
I try to make it opaque by using handle.
in FPC THandle = TSocket = Longint while in Delphi is Cardinal.

to make me understand, I try change some data type


  TEPollData = record
    case Integer of
      0: (Ptr      :Pointer);
      1: (Handle  :THandle);      // make it opaque
      2: (U32      :UInt32 );      
      3: (U64      :UInt64 );
  end;


from http://man7.org/linux/man-pages/man2/epoll_ctl.2.html

   The events member is a bit mask composed by ORing together zero or
   more of the following available event types:


   TEPollEventTypes = UInt32;

  TEPollEvent = packed record
    Types        :TEPollEventTypes;      // for me to distinguish from
list
    Data         :TEPollData;
  end;
   PEPollEvent = ^TEPollEvent;


so, event consist of which handle to be notify by event types
like EPOLLIN for read and EPOLLOUT for write

   Event.Types           (EPOLLIN or EPOLLOUT ...)
   Event.Data.Handle


   // event list is array of event
  TEPollEventList = array of TEPollEvent;

  function epoll_wait(
    epfd        :THandle;
    events      :@EPollEventList[0];
    maxevents   :Int32;
    timeout     :Int32
  ) :cint;



so I split to two unit zepoll.pas and zserver.pas

from your code seem your listenSocket just use EPOLLIN, same as my
first example,
assuming using single thread mode, may be its ok

as in zserver I also simulate long process
now the problem arise, it would block all.

for test make connection with two telnet session
one type timeout the other just do echo.


I m searching epoll because limitation of select

refer to https://medium.com/@copyconstruct/the-method-to-epolls-
madness-d9d2d6378642

seem another problem arise

https://idea.popcount.org/2017-02-20-epoll-is-fundamentally-broken-12/
https://idea.popcount.org/2017-03-20-epoll-is-fundamentally-broken-22/

result of epoll_wait
epoll_wait( 0)      server immediately timeout not have chances to
accept
epoll_wait(-1)     never timeout
epoll_wait(1000)  if busy never timeout


maybe I'm not clear enough about accept timeout, what I mean is
after accept if there is no data sent at period of time, the socket
should be close.


https://www.winsocketdotnetworkprogramming.com/winsock2programming/win
sock2advancedsocketoptionioctl7.html

SO_CONNECT_TIME is a Microsoft-specific option that returns
the number of seconds a connection has been established.

A malicious application could perform a denial of
service attack by making many connections without sending data.

currently linux dont have that option.

about thundering herd
https://idea.popcount.org/2017-01-06-select-is-fundamentally-broken/

I cannot say enough, it is like thundering gru, and all his minion
attack him that make him stunned.

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i unit
zEpoll;{$mode objfpc}{$h+}interface uses BaseUnix, Linux; type  
TEPollData = record    case Integer of      0: (Ptr     :Pointer);   
 1: (Handle  :THandle);      2: (U32     :UInt32 );      3: (U64    
:UInt64 );  end;   TEPollEventTypes = UInt32;   TEPollEvent = packed
record    Types       :TEPollEventTypes;    Data        :TEPollData; 
end;   TEPollEventList = array of TEPollEvent;   TEPoll = class
(TObject)  public    const      TYPE_IN           :TEPollEventTypes =
$00000001;      TYPE_PRI          :TEPollEventTypes = $00000002;     
TYPE_OUT          :TEPollEventTypes = $00000004;      TYPE_ERR       
 :TEPollEventTypes = $00000008;      TYPE_HUP        
 :TEPollEventTypes = $00000010;      TYPE_RD_NORM    
 :TEPollEventTypes = $00000040;      TYPE_RD_BAND    
 :TEPollEventTypes = $00000080;      TYPE_WR_NORM    
 :TEPollEventTypes = $00000100;      TYPE_WR_BAND    
 :TEPollEventTypes = $00000200;      TYPE_MSG        
 :TEPollEventTypes = $00000400;      TYPE_RD_HUP      
:TEPollEventTypes = $00002000;      TYPE_EXCLUSIVE  
 :TEPollEventTypes = $10000000;      TYPE_WAKE_UP    
 :TEPollEventTypes = $20000000;      TYPE_ONE_SHOT    
:TEPollEventTypes = $40000000;      TYPE_EDGE_TRIGGER
:TEPollEventTypes = $80000000;      TIMEOUT   :Int32 =  0;      ERROR
   :Int32 = -1;  private    FMaxEvent   :Int32;    FEPoll    
 :THandle;    FEventList  :TEPollEventList;  public    constructor
Create(const AMaxEvent :Int32);    destructor  Destroy; override;   
function    Add       (const AHandle  :THandle;     const ATypes
:TEPollEventTypes) :TError;    function    Modify    (const AHandle
 :THandle;     const ATypes :TEPollEventTypes) :TError;    function  
 Delete    (const AHandle  :THandle;     const ATypes
:TEPollEventTypes) :TError;    function    IsType    (const AEvent  
:TEPollEvent; const ATypes :TEPollEventTypes) :Boolean;    function  
 Wait      (const ATimeout :Int32) :Int32;    property    EventList
:TEPollEventList read FEventList;  end; implementation constructor
TEPoll.Create(const AMaxEvent :Int32);  begin    inherited Create;   
FMaxEvent := AMaxEvent;    SetLength(FEventList,  FMaxEvent);   
FEPoll := epoll_create(FMaxEvent);  end; destructor TEPoll.Destroy; 
begin    fpClose(FEPoll);    inherited Destroy;  end; function
TEPoll.Add(const AHandle :THandle; const ATypes :TEPollEventTypes)
:TError;  var event :TEPollEvent;  begin    event.Data.Handle :=
AHandle;    event.Types       := ATypes;    result :=
epoll_ctl(FEPoll, EPOLL_CTL_ADD, event.Data.Handle, @event); 
end; function TEPoll.Modify(const AHandle :THandle; const ATypes
:TEPollEventTypes) :TError;  var event :TEPollEvent;  begin   
event.Data.Handle := AHandle;    event.Types       := ATypes;   
result := epoll_ctl(FEPoll, EPOLL_CTL_MOD, event.Data.Handle,
@event);  end; function TEPoll.Delete(const AHandle :THandle; const
ATypes :TEPollEventTypes) :TError;  var event :TEPollEvent;  begin   
event.Data.Handle := AHandle;    event.Types       := ATypes;   
result := epoll_ctl(FEPoll, EPOLL_CTL_DEL, event.Data.Handle,
@event);  end; function TEPoll.IsType(const AEvent :TEPollEvent; const
ATypes :TEPollEventTypes) :Boolean;  begin    result := ((AEvent.Types
and ATypes) = ATypes);  end; function    TEPoll.Wait(const ATimeout
:Int32) :Int32;  begin    result := epoll_wait(FEPoll, @FEventList[0],
FMaxEvent, ATimeout);  end; end.  
server

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i unit
zServer;{$mode objfpc}{$h+}interface uses BaseUnix, Termio, Sockets,
zEPoll, Errors, SysUtils; type   TServer = class (TObject)  private  
FEPoll      :TEPoll;    FServer     :TSocket;    FPort       :UInt32; 
  FError      :TError;  public    constructor Create(const APort
:UInt32);    destructor  Destroy; override;    function  
 Disconnect(const ASocket :TSocket) :TError;    function    Receive  
(const ASocket :TSocket; var   ABuffer :String) :TError;    function 
 Send      (const ASocket :TSocket; const ABuffer :String) :TError;  
procedure   Listen;    property    Error :TError read FError; 
end; implementation constructor TServer.Create(const APort :UInt32); 
begin    inherited Create;    FPort  := APort;    FEPoll :=
TEPoll.Create(128);  end; destructor  TServer.Destroy;  begin   
FEPoll.Free;    inherited Destroy;  end; function  
 TServer.Disconnect(const ASocket :TSocket) :TError;  begin    {     
from https://idea.popcount.org/2017-03-20-epoll-is-fundamentally-
broken-22/      Remember this advice:      You always must always
explicitly call epoll_ctl(EPOLL_CTL_DEL) before calling close().    } 
  FEPoll.Delete(ASocket, TEPoll.TYPE_IN);    fpShutdown(ASocket,
SHUT_WR);    fpClose(ASocket);    result := SocketError; 
end; function    TServer.Receive(const ASocket :TSocket; var ABuffer
:String) :TError;  var readlen :Int32;  begin    ABuffer := '';    if
fpIOCtl(ASocket, FIONREAD, @readlen) = 0 then      begin       
SetLength(ABuffer, readlen);        fpRecv(ASocket, PChar(ABuffer),
readlen, 0);      end;    result := SocketError;    WriteLn('rx[',
result, ']: ', ABuffer);  end; function    TServer.Send(const ASocket
:TSocket; const ABuffer :String) :TError;  begin    fpSend(ASocket,
PChar(ABuffer), Length(ABuffer), 4);    result := SocketError;   
WriteLn('tx[', result, ']: ', ABuffer);  end; procedure  
TServer.Listen;  var    i, WaitReturn,    addrSize :Int32;    event  
 :TEPollEvent;    client   :TSocket;    addr     :TInetSockAddr;   
msg      :String;  begin     FServer := fpSocket(AF_INET, SOCK_STREAM,
0);     if FServer = -1 then      begin        FError := SocketError; 
      Exit;      end;     addr.sin_family       := AF_INET;   
addr.sin_port         := NToHS(FPort);    addr.sin_addr.s_addr  := 0; 
  addrSize              := SizeOf(addr);     if fpBind(FServer, @addr,
addrSize) <> 0 then      begin        FError := SocketError;       
Exit;      end;     if fpListen(FServer, 128) <> 0 then      begin   
   FError := SocketError;        Exit;      end;     // thundering
herd ???    if FEPoll.Add(FServer, TEPoll.TYPE_IN +
TEPoll.TYPE_EXCLUSIVE) = TEPoll.ERROR then      begin        FError :=
fpGetErrNo;        Exit;      end;     FError := 0;    repeat      
WaitReturn := FEPoll.Wait(-1); // wait for 1 second       // as
zamronypj mention      if WaitReturn = TEPoll.ERROR then       
WriteLn('wait error')       else if WaitReturn = TEPoll.TIMEOUT then 
     WriteLn('wait timeout')       else if WaitReturn > 0 then  // n
event ready list        begin           for i:=0 to WaitReturn-1 do  
        begin              event  := FEPoll.EventList[i];            
// check for event              Write('handle: ', event.Data.Handle, '
event types: ', event.Types);              if FEPoll.IsType(event,
TEPoll.TYPE_IN          ) then Write(' EPOLLIN');              if
FEPoll.IsType(event, TEPoll.TYPE_PRI         ) then Write('
EPOLLPRI');              if FEPoll.IsType(event, TEPoll.TYPE_OUT     
  ) then Write(' EPOLLOUT');              if FEPoll.IsType(event,
TEPoll.TYPE_ERR         ) then Write(' EPOLLERR');              if
FEPoll.IsType(event, TEPoll.TYPE_HUP         ) then Write('
EPOLLHUP');              if FEPoll.IsType(event, TEPoll.TYPE_RD_NORM 
  ) then Write(' EPOLLRDNORM');              if FEPoll.IsType(event,
TEPoll.TYPE_RD_BAND     ) then Write(' EPOLLRDBAND');              if
FEPoll.IsType(event, TEPoll.TYPE_WR_NORM     ) then Write('
EPOLLWRNORM');              if FEPoll.IsType(event,
TEPoll.TYPE_WR_BAND     ) then Write(' EPOLLWRBAND');              if
FEPoll.IsType(event, TEPoll.TYPE_MSG         ) then Write('
EPOLLMSG');              if FEPoll.IsType(event, TEPoll.TYPE_RD_HUP  
  ) then Write(' EPOLLRDHUP');              if FEPoll.IsType(event,
TEPoll.TYPE_EXCLUSIVE   ) then Write(' EPOLLEXCLUSIVE');             
if FEPoll.IsType(event, TEPoll.TYPE_WAKE_UP     ) then Write('
EPOLLWAKEUP');              if FEPoll.IsType(event,
TEPoll.TYPE_ONE_SHOT    ) then Write(' EPOLLONESHOT');              if
FEPoll.IsType(event, TEPoll.TYPE_EDGE_TRIGGER) then Write('
EPOLLET');              WriteLn;               // how to test ERR and
HUP ??              if   FEPoll.IsType(event, TEPoll.TYPE_ERR)       
       or FEPoll.IsType(event, TEPoll.TYPE_HUP)                or
FEPoll.IsType(event, TEPoll.TYPE_RD_HUP) then                 
self.Disconnect(event.Data.Handle)              else               
begin                  if event.Data.Handle = FServer then           
       begin                      client := fpAccept(FServer, @addr,
@addrSize);                      if client <> -1 then                
      FEPoll.Add(client, TEPoll.TYPE_IN + TEPoll.TYPE_ERR +
TEPoll.TYPE_HUP + TEPoll.TYPE_RD_HUP);                    end        
        else                    begin                      if
FEPoll.IsType(event, TEPoll.TYPE_IN) then                       
begin                          self.Receive(event.Data.Handle, msg);  
                        // simulate long process                     
   if Pos('timeout', msg) > 0 then                            begin  
                          WriteLn('start sleep');                    
        Sleep(20000); // long process would block all                
            WriteLn('stop  sleep');                            end;  
                       self.Send(event.Data.Handle, msg);            
             if Pos('quit', msg) > 0 then                           
self.Disconnect(event.Data.Handle);                           if
Pos('shutdown', msg) > 0 then                            FError :=
-1;                         end;                    end;             
 end;            end;        end;    until FError <> 0;    
WriteLn('error : ', FError, ' - ', StrError(FError));   
self.Disconnect(FServer);  end; end.  
main

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i program
epollserver;{$mode objfpc}{$h+}uses CMem, CThreads, zServer; var
server :TServer; begin  server := TServer.Create(8000); 
server.Listen;  server.Free;end.

----------------------------------------------------------------------
zamronypj · 2019-08-01 22:18

CMIIW, if you need to check if client connection has been idle for too
long, you can make client connection non-blocking after you accept
connection.

On Linux, accept() implementation does not inherit non-blocking status
of listening socket

Read
http://man7.org/linux/man-pages/man2/accept.2.html

on "Conforming To" section.

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i
clientSocket := fpAccept(fServer, @addr,
@addrSize);setNonBlocking(clientSocket);
So next time, clientSocket is signaled, recv() call will not block,
and you can check how many time has elapsed since last recv() call,
then you can decide, if it should be closed or not.

Also you can set client socket recv() timeout via setsockopt() after
you accept connection and set SO_RCVTIMEO option to timeout value that
you need.

http://man7.org/linux/man-pages/man2/setsockopt.2.html

According to

http://man7.org/linux/man-pages/man2/recv.2.html


if recv() returns 0 you can assume client finish sending data, so it
can be closed.

If use non blocking socket, when it returns -1 and SocketError()
returns ESysEAGAIN or ESysEWOULDBLOCK you should retry recv() again
next time it's signaled.

before looping again to call epoll_wait(), I think it is better to
check all client connection that has been idle too long which is
client connection which exceed timeout value threshold after being
accepted or received , and just close it and remove it from epoll
interest list.