#############################################################################
##
## iohub.gd GAP 4 package IO
## Max Neunhoeffer
##
## Copyright (C) by Max Neunhoeffer
## This file is free software, see license information at the end.
##
## This file contains functions for a generic client server framework
## for GAP
##
## Main points:
##
## - handle multiple connections using IO multiplexing
## - single threaded
## - use pickling for data transfer
##
InstallMethod( IOHub, "constructor without arguments", [ ],
function( )
local s;
s := rec( sock := fail, inqueue := [], outqueue := [],
tosend := [], torecv := [], inbuf := [], outbuf := [],
connections := [], isactive := true ) ;
Objectify(IOHubType, s);
return s;
end );
InstallMethod( AttachServingSocket, "for an address and a port",
[ IsIOHub, IsStringRep, IsPosInt ],
function( s, address, port )
s!.sock := IO_socket(IO.PF_INET,IO.SOCK_STREAM,"tcp");
if s!.sock = fail then return fail; fi;
IO_setsockopt(s!.sock,IO.SOL_SOCKET,IO.SO_REUSEADDR,"\001\001\001\001");
if IO_bind(s!.sock,IO_MakeIPAddressPort(address,port)) = fail then
IO_close(s!.sock);
s!.sock := fail;
return fail;
fi;
if not(IO_listen(s!.sock,5)) then
IO_close(s!.sock);
s!.sock := fail;
return fail;
fi;
return s!.sock;
end );
InstallMethod( CloseConnection, "for an IO hub and a positive integer",
[ IsIOHub, IsPosInt ],
function( s, nr )
# First remove all entries in the queue from or for this connection:
local i;
if not(IsBound(s!.connections[nr])) then
return fail;
fi;
if IsBound(s!.connections[nr][1]) then
i := 1;
while i <= Length(s!.inqueue) do
if s!.inqueue[i][1] = nr then
Remove(s!.inqueue[i]);
else
i := i + 1;
fi;
od;
IO_close(s!.connections[nr][1]);
fi;
if IsBound(s!.connections[nr][2]) then
i := 1;
while i <= Length(s!.outqueue) do
if s!.outqueue[i][1] = nr then
Remove(s!.outqueue[i]);
else
i := i + 1;
fi;
od;
IO_close(s!.connections[nr][2]);
fi;
Unbind(s!.connections[nr]);
Unbind(s!.tosend[nr]);
Unbind(s!.torecv[nr]);
Unbind(s!.inbuf[nr]);
Unbind(s!.outbuf[nr]);
Print("Connection #",nr," closed.\n");
end );
InstallMethod( ShutdownServingSocket, "for an IO hub",
[IsIOHub],
function(s)
if s!.sock <> fail then
IO_close(s!.sock);
s!.sock := fail;
fi;
end );
InstallMethod( Shutdown, "for an IO hub",
[ IsIOHub ],
function( s )
local i;
if not(s!.isactive) then return; fi;
for i in [1..Length(s!.connections)] do
CloseConnection(s,i);
od;
ShutdownServingSocket(s);
s!.isactive := false;
end );
InstallMethod( ViewObj, "for a tcp server",
[ IsIOHub ],
function( s )
local nr;
if s!.isactive then
Print("<IO hub");
if s!.sock <> fail then
Print(" with serving socket");
fi;
nr := Number([1..Length(s!.connections)],
i->IsBound(s!.connections[i]) and
IsBound(s!.connections[i][1]));
if nr > 0 then
Print(", reading from ",nr," fds");
fi;
nr := Number([1..Length(s!.connections)],
i->IsBound(s!.connections[i]) and
IsBound(s!.connections[i][2]));
if nr > 0 then
Print(", writing to ",nr," fds");
fi;
Print(">");
else
Print("<IO hub already shut down>");
fi;
end );
InstallMethod( NewConnection, "for an IO hub and two integers",
[ IsIOHub, IsInt, IsInt ],
function( s, inp, out )
local i,l;
if not(s!.isactive) then return fail; fi;
i := Length(s!.connections)+1; # do not reuse old connection numbers
l := [];
if inp > 0 then l[1] := inp; fi;
if out > 0 then l[2] := out; fi;
s!.connections[i] := l;
s!.tosend[i] := 0;
s!.torecv[i] := 0;
s!.inbuf[i] := EmptyString(8);
s!.outbuf[i] := "";
return i;
end );
InstallMethod( NewTCPConnection, "for an IO hub, an address and a port",
[ IsIOHub, IsStringRep, IsPosInt ],
function( s, address, port )
local t;
t := IO_socket(IO.PF_INET,IO.SOCK_STREAM,"tcp");
if IO_connect(t,IO_MakeIPAddressPort(address,port)) = fail then
IO_close(t);
return fail;
fi;
return NewConnection(s,t,t);
end );
InstallMethod( AcceptNewConnection, "for an IO hub",
[ IsIOHub ],
function( s )
local t,i;
if not(s!.isactive) or not(IsBound(s!.sock)) then
return fail;
fi;
t := IO_accept(s!.sock,IO_MakeIPAddressPort("0.0.0.0",0));
i := NewConnection( s, t, t );
Print("Got new connection #",i,"...\n");
return i;
end );
InstallMethod( GetInput, "for an IO hub and an integer",
[ IsIOHub, IsInt ],
function( s, i )
local p;
if not(s!.isactive) then return fail; fi;
if i = 0 then # get something from any connection
if Length(s!.inqueue) = 0 then
return false;
else
return Remove(s!.inqueue,1);
fi;
else
p := 1;
while p <= Length(s!.inqueue) and s!.inqueue[p][1] <> i do
p := p + 1;
od;
if p > Length(s!.inqueue) then
return false;
else
return Remove(s!.inqueue,p);
fi;
fi;
end );
InstallMethod( SubmitOutput, "for an IO hub, a positive integers and an obj",
[ IsIOHub, IsPosInt, IsStringRep ],
function( s, i, o )
if not(IsBound(s!.connections[i]) and IsBound(s!.connections[i][2])) then
Error("This connection is closed or has no output");
return fail;
fi;
Add(s!.outqueue,[i,o]);
return true;
end );
InstallMethod( OutputQueue, "for an IO hub",
[ IsIOHub ],
function( s ) return s!.outqueue; end );
InstallMethod( InputQueue, "for an IO hub",
[ IsIOHub ],
function( s ) return s!.inqueue; end );
InstallMethod( StoreLenIn8Bytes, "for a string and a len",
[IsStringRep, IsInt],
function( st, len )
local c,i;
for i in [1..8] do
c := len mod 256;
st[i] := CHAR_INT(c);
len := (len - c) / 256;
od;
end );
InstallMethod( GetLenFrom8Bytes, "for a string",
[IsStringRep],
function( st )
local len,i;
len := 0;
for i in [8,7..1] do
len := len * 256 + INT_CHAR(st[i]);
od;
return len;
end );
InstallMethod( DoIO, "for an IO hub", [ IsIOHub ],
function( s ) return DoIO(s,false); end );
InstallMethod( DoIO, "for an IO hub and a boolean",
[ IsIOHub, IsBool ],
function( s, block )
# This uses select to see to all open connections including the
# original socket to perform all possible IO on them. New connections
# are created if needed and those to which network connectivity is
# lost are closed.
# Note that this does not automatically call the worker on the input
# queue.
# However, it does serve the output queue.
local activity,bytes,hadactivity,i,infds,inptab,j,len,nr,outfds,outtab,st;
if not(s!.isactive) then return fail; fi;
hadactivity := false;
repeat
activity := false;
# First we check whether some output from the queue has to be sent:
j := 1;
while j <= Length(s!.outqueue) do
i := s!.outqueue[j][1];
if s!.tosend[i] = 0 then # idle
st := Concatenation("00000000",s!.outqueue[j][2]);
# the first 8 will be the length
len := Length(st);
StoreLenIn8Bytes(st,len-8);
s!.outbuf[i] := st;
s!.tosend[i] := len;
Remove(s!.outqueue,j);
else
j := j + 1;
fi;
od;
# Now do a select:
infds := EmptyPlist(Length(s!.connections)+1);
outfds := EmptyPlist(Length(s!.connections));
inptab := EmptyPlist(Length(s!.connections)+1);
outtab := EmptyPlist(Length(s!.connections));
for i in [1..Length(s!.connections)] do
if IsBound(s!.connections[i]) then
if IsBound(s!.connections[i][1]) then
Add(infds,s!.connections[i][1]);
Add(inptab,i);
fi;
if IsBound(s!.connections[i][2]) and s!.tosend[i] <> 0 then
Add(outfds,s!.connections[i][2]);
Add(outtab,i);
fi;
fi;
od;
if s!.sock <> fail then
Add(infds,s!.sock);
Add(inptab,0);
fi;
if block and not(hadactivity) then
nr := IO_select(infds,outfds,[],false,false);
else
nr := IO_select(infds,outfds,[],0,0);
fi;
if nr > 0 then
# Look for possible output first:
for j in [1..Length(outfds)] do
if outfds[j] <> fail then
activity := true;
i := outtab[j];
bytes := IO_write(s!.connections[i][2],s!.outbuf[i],
Length(s!.outbuf[i])-s!.tosend[i],
s!.tosend[i]);
if bytes <= 0 then # an error
CloseConnection(s,i);
# maybe we want to have a callback here!
else
s!.tosend[i] := s!.tosend[i] - bytes;
if s!.tosend[i] = 0 then
Unbind(s!.outbuf[i]);
fi;
fi;
fi;
od;
# Now look for possible inputs next:
# We need to remember that some connections might already
# me closed by the output routine above!
for j in [1..Length(infds)] do
if infds[j] <> fail then
activity := true;
i := inptab[j];
if i = 0 then
AcceptNewConnection(s);
else
if IsBound(s!.connections[i]) then
if s!.torecv[i] = 0 then # read length
bytes := IO_read(s!.connections[i][1],
s!.inbuf[i],Length(s!.inbuf[i]),
8-Length(s!.inbuf[i]));
if bytes <= 0 then # an error
CloseConnection(s,i);
# maybe we want to have a callback here!
continue;
fi;
if Length(s!.inbuf[i]) = 8 then
s!.torecv[i]:=GetLenFrom8Bytes(s!.inbuf[i]);
s!.inbuf[i] := EmptyString(s!.torecv[i]);
fi;
else # we are in the reading process
bytes := IO_read(s!.connections[i][1],
s!.inbuf[i],Length(s!.inbuf[i]),
s!.torecv[i]-Length(s!.inbuf[i]));
if bytes <= 0 then # an error
CloseConnection(s,i);
# maybe we want to have a callback here!
continue;
fi;
if Length(s!.inbuf[i]) = s!.torecv[i] then
Add(s!.inqueue,[i,s!.inbuf[i]]);
s!.torecv[i] := 0;
s!.inbuf[i] := EmptyString(8);
fi;
fi;
fi;
fi;
fi;
od;
fi;
if activity then hadactivity := true; fi;
until activity = false;
return hadactivity;
end );
##
## This program 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.
##
## This program 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 this program. If not, see <
https://www.gnu.org/licenses/>.
##
