[libt2n] / src / socket_server.cpp

/***************************************************************************
 *   Copyright (C) 2006 by Gerd v. Egidy                                   *
 *   gve@intra2net.com                                                     *
 *                                                                         *
 *   This library is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Lesser General Public License version   *
 *   2.1 as published by the Free Software Foundation.                     *
 *                                                                         *
 *   This library 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 Lesser General Public License for more details.                   *
 *                                                                         *
 *   You should have received a copy of the GNU Lesser General Public      *
 *   License along with this program; if not, write to the                 *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/

#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <netdb.h>
#include <fcntl.h>
#include <time.h>
#include <pwd.h>
#include <grp.h>

#include <sstream>

#include "socket_server.hxx"
#include "t2n_exception.hxx"

using namespace std;

namespace libt2n
{

socket_server::socket_server(int port, const char* ip)
    : server()
{
    socket_type=tcp_s;

}

socket_server::socket_server(const char* path, mode_t filemode, const char* user, const char* group)
    : server()
{
    socket_type=unix_s;
    unix_path=path;

    /* Create the socket. */
    sock = socket (PF_UNIX, SOCK_STREAM, 0);
    if (sock < 0)
    {
        string err="error opening socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    set_socket_options(sock);

    /* Give the socket a name. */
    struct sockaddr_un unix_name;
    unix_name.sun_family = AF_UNIX;
    strncpy (unix_name.sun_path, unix_path.c_str(),sizeof(unix_name.sun_path));

    /* just to make sure there is no other socket file */
    unlink (unix_name.sun_path);

    if (bind (sock, (struct sockaddr *) &unix_name, sizeof (unix_name)) < 0)
    {
        string err="error binding socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    /* change permissions */
    if (chmod (unix_name.sun_path, filemode) != 0) 
    {
        string err="error changing permission: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    struct passwd *socket_user = getpwnam (user);
    if (socket_user == NULL) 
    {
        string err="error getting socket user: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    struct group *socket_group = getgrnam (group);
    if (socket_group == NULL) 
    {
        string err="error getting socket group: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    if (chown (unix_name.sun_path, socket_user->pw_uid, socket_group->gr_gid) != 0) 
    {
        string err="error changing socket ownership: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    if (listen (sock, 5) < 0)
    {
        string err="error listening to socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    /* clear & insert server sock into the fd_tab to prepare select */
    FD_ZERO(&connection_set);
    FD_SET (sock, &connection_set);
}

void socket_server::set_socket_options(int sock)
{
    int i=1;

    /* fast reuse enable */
    if (setsockopt(sock,SOL_SOCKET, SO_REUSEADDR, &i, sizeof(i)) < 0)
    {
        string err="error setting socket option: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    /* keepalive enable */
    if (setsockopt(sock,SOL_SOCKET, SO_KEEPALIVE, &i, sizeof(i)) < 0)
    {
        string err="error setting socket option: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    /* close on exec */
    int fdflags;
    fdflags=fcntl(sock,F_GETFD, 0);
    if (fdflags < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }
    fdflags |= FD_CLOEXEC;
    if (fcntl(sock,F_SETFD,fdflags) < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    /* non-blocking mode */
    int flflags;
    flflags=fcntl(sock,F_GETFL,0);
    if (flflags < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }
    flflags |= O_NONBLOCK;
    if (fcntl(sock,F_SETFL,flflags) < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }
}

socket_server::~socket_server()
{
    close(sock);

    if (socket_type==unix_s)
        unlink(unix_path.c_str());
}

void socket_server::new_connection()
{
    struct sockaddr_un clientname;

    unsigned int size = sizeof (clientname);
    int newsock = accept (sock,(struct sockaddr *) &clientname,&size);
    if (newsock < 0)
    {
        if (errno == EAGAIN)
        {
            log(error, "accept error (EAGAIN): no connection waiting");
            return;
        }

        /* default: break */
        string err="error accepting connection: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    FD_SET (newsock, &connection_set);

    int i=1;

    /* keepalive enable */
    if (setsockopt(newsock,SOL_SOCKET, SO_KEEPALIVE, &i, sizeof(i)) < 0)
    {
        string err="error setting socket option: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    /* close on exec */
    int fdflags;
    fdflags=fcntl(newsock,F_GETFD, 0);
    if (fdflags < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }
    fdflags |= FD_CLOEXEC;
    if (fcntl(newsock,F_SETFD,fdflags) < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    /* non-blocking mode */
    int flflags;
    flflags=fcntl(newsock,F_GETFL,0);
    if (flflags < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }
    flflags |= O_NONBLOCK;
    if (fcntl(newsock,F_SETFL,flflags) < 0)
    {
        string err="fcntl error on socket: ";
        err+=strerror(errno);
        log(error, err);
        throw t2n_server_error(err);
    }

    add_connection(new socket_connection(newsock, get_default_timeout()));

    return;
}

void socket_server::fill_buffer(long long usec_timeout)
{
    fd_set used_fdset=connection_set;

    /* set timeout */
    struct timeval tval;
    struct timeval *timeout_ptr;

    if (usec_timeout == -1)
        timeout_ptr = NULL;
    else
    {
        timeout_ptr = &tval;

        // timeout von long long usec in int sec + int usec umrechnen
        tval.tv_sec = usec_timeout / 1000000;
        tval.tv_usec = usec_timeout % 1000000;
    }

    int ret=select (FD_SETSIZE, &used_fdset, NULL, NULL, timeout_ptr);

    if (ret < 0)
    {
        if (errno == EINTR)
        {
            // select interrupted by signal
            ret=0;
        }
        else
        {
            string err="select error: ";
            err+=strerror(errno);
            log(error, err);
            throw t2n_server_error(err);
        }
    }

    if (ret > 0)
    {
        // we have data pending

        // check for new connection
        if (FD_ISSET (sock, &used_fdset))
        {
            new_connection();
        }

        // check all connections for pending data
        fill_connection_buffers();
    }

    return;
}

void socket_server::fill_connection_buffers()
{
    std::map<unsigned int, connection*>::iterator ie=connections.end();
    for(std::map<unsigned int, connection*>::iterator i=connections.begin(); i != ie; i++)
        if (!i->second->is_closed())
        {
            socket_connection* cp=dynamic_cast<socket_connection*>(i->second);
            cp->fill_buffer(connection_set);
        }
}

socket_connection::socket_connection(int _sock, int _timeout)
    : connection(_timeout)
{
    sock=_sock;
}

void socket_connection::close()
{

}

void socket_connection::fill_buffer(fd_set &cur_fdset)
{
    bool try_again=false;

    if (is_closed() || !FD_ISSET (sock, &cur_fdset))
        return;                                                // no data pending or connection closed

    // data pending -> go and get it
    char socket_buffer[recv_buffer_size];

    int nbytes = read (sock, socket_buffer, recv_buffer_size);
    if (nbytes < 0)
    {
        if (errno == EAGAIN)
        {
            if (my_server)
                my_server->log(server::error,"read error: no data (EAGAIN) for connection "+get_id_string());
            return;
        }
        else if (errno == EINTR)
        {
            // interrupted, try again
            try_again=true;
        }
        else
        {
            if (my_server)
                my_server->log(server::error,"error reading from socket of connection "+get_id_string()+": "+strerror(errno));
            return;
        }
    }

    // End-of-file
    if (nbytes == 0 && !try_again)
    {
        close();
        return;
    }

    // Data read -> store it
    if (nbytes > 0)
        buffer.assign(socket_buffer,nbytes);

    // more data waiting?
    fd_set active_fd_set;
    struct timeval tval;

    FD_ZERO (&active_fd_set);
    FD_SET (sock, &active_fd_set);

    /* no waiting */
    tval.tv_sec=0;
    tval.tv_usec=0;

    if (select (FD_SETSIZE, &active_fd_set, NULL, NULL, &tval) >0)
    {
        /* more data waiting -> recurse */
        fill_buffer(active_fd_set);
    }

    reset_timeout();

    return;
}

void socket_connection::write(const std::string& data)
{
    static const unsigned int write_block_size=4096;

    if (is_closed())
        return;

    // prepend packet size to data
    packet_size_indicator psize=data.size();
    string send_data(data);
    send_data.insert(0,(char*)psize,sizeof(packet_size_indicator));

    int offset = 0;
    while (offset < send_data.size())
    {
        unsigned int write_size=write_block_size;

        if (offset+write_size > send_data.size())
            write_size = send_data.size()-offset;

        int rtn;
        while ((rtn=::write(sock, send_data.data()+offset, write_size)) &&
               rtn == -1 && (errno == EAGAIN || errno == EINTR))
        {
            usleep (80000);
            if (my_server)
                my_server->log(server::debug,"resuming write() call after EAGAIN or EINTR for connection "+get_id_string());
        }

        if (rtn == -1)
        {
            if (my_server)
                my_server->log(server::error,"write() error on connection "+get_id_string()+": "+strerror(errno));
        }
        else if (rtn != write_size)
        {
            if (my_server)
            {
                ostringstream msg;
                msg << "write() error on connection " << get_id() 
                    << ": wrote " << rtn << " bytes, should have been " 
                    << write_size << " (complete: " << send_data.size() << ")";

                my_server->log(server::error,msg.str());
            }
        }

        offset += write_size;
    }

    reset_timeout();

    return;
}

}
Commit	Line	Data
ac7fdc22 GE	1	/***************************************************************************
	2	* Copyright (C) 2006 by Gerd v. Egidy *
	3	* gve@intra2net.com *
	4	* *
	5	* This library is free software; you can redistribute it and/or modify *
	6	* it under the terms of the GNU Lesser General Public License version *
	7	* 2.1 as published by the Free Software Foundation. *
	8	* *
	9	* This library is distributed in the hope that it will be useful, *
	10	* but WITHOUT ANY WARRANTY; without even the implied warranty of *
	11	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
	12	* GNU Lesser General Public License for more details. *
	13	* *
	14	* You should have received a copy of the GNU Lesser General Public *
	15	* License along with this program; if not, write to the *
	16	* Free Software Foundation, Inc., *
	17	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
	18	***************************************************************************/
	19
0cf4dc9b GE	20	#include <stdio.h>
	21	#include <errno.h>
	22	#include <stdlib.h>
	23	#include <unistd.h>
	24	#include <sys/types.h>
	25	#include <sys/socket.h>
	26	#include <sys/un.h>
	27	#include <sys/time.h>
	28	#include <arpa/inet.h>
	29	#include <netinet/in.h>
	30	#include <netdb.h>
	31	#include <fcntl.h>
	32	#include <time.h>
	33	#include <pwd.h>
	34	#include <grp.h>
	35
aa499d20 GE	36	#include <sstream>
aa499d20 GE	37
ac7fdc22	38	#include "socket_server.hxx"
0cf4dc9b GE	39	#include "t2n_exception.hxx"
	40
	41	using namespace std;
ac7fdc22 GE	42
	43	namespace libt2n
	44	{
	45
	46	socket_server::socket_server(int port, const char* ip)
	47	: server()
	48	{
0cf4dc9b	49	socket_type=tcp_s;
ac7fdc22 GE	50
	51	}
	52
0cf4dc9b	53	socket_server::socket_server(const char* path, mode_t filemode, const char* user, const char* group)
ac7fdc22 GE	54	: server()
ac7fdc22 GE	55	{
0cf4dc9b	56	socket_type=unix_s;
ac7fdc22 GE	57	unix_path=path;
ac7fdc22 GE	58
ac7fdc22	59	/* Create the socket. */
0cf4dc9b GE	60	sock = socket (PF_UNIX, SOCK_STREAM, 0);
0cf4dc9b GE	61	if (sock < 0)
ac7fdc22 GE	62	{
ac7fdc22 GE	63	string err="error opening socket: ";
0cf4dc9b GE	64	err+=strerror(errno);
	65	log(error, err);
	66	throw t2n_server_error(err);
	67	}
	68
	69	set_socket_options(sock);
	70
	71	/* Give the socket a name. */
	72	struct sockaddr_un unix_name;
	73	unix_name.sun_family = AF_UNIX;
	74	strncpy (unix_name.sun_path, unix_path.c_str(),sizeof(unix_name.sun_path));
	75
	76	/* just to make sure there is no other socket file */
	77	unlink (unix_name.sun_path);
	78
	79	if (bind (sock, (struct sockaddr *) &unix_name, sizeof (unix_name)) < 0)
	80	{
	81	string err="error binding socket: ";
	82	err+=strerror(errno);
	83	log(error, err);
	84	throw t2n_server_error(err);
	85	}
	86
	87	/* change permissions */
	88	if (chmod (unix_name.sun_path, filemode) != 0)
	89	{
	90	string err="error changing permission: ";
	91	err+=strerror(errno);
	92	log(error, err);
	93	throw t2n_server_error(err);
	94	}
	95
	96	struct passwd *socket_user = getpwnam (user);
	97	if (socket_user == NULL)
	98	{
	99	string err="error getting socket user: ";
	100	err+=strerror(errno);
	101	log(error, err);
	102	throw t2n_server_error(err);
	103	}
	104
	105	struct group *socket_group = getgrnam (group);
	106	if (socket_group == NULL)
	107	{
	108	string err="error getting socket group: ";
	109	err+=strerror(errno);
	110	log(error, err);
	111	throw t2n_server_error(err);
	112	}
	113
	114	if (chown (unix_name.sun_path, socket_user->pw_uid, socket_group->gr_gid) != 0)
	115	{
	116	string err="error changing socket ownership: ";
	117	err+=strerror(errno);
	118	log(error, err);
	119	throw t2n_server_error(err);
	120	}
	121
	122	if (listen (sock, 5) < 0)
	123	{
	124	string err="error listening to socket: ";
	125	err+=strerror(errno);
	126	log(error, err);
	127	throw t2n_server_error(err);
128	}
129
130	/* clear & insert server sock into the fd_tab to prepare select */
131	FD_ZERO(&connection_set);
132	FD_SET (sock, &connection_set);
133	}
134
135	void socket_server::set_socket_options(int sock)
136	{
137	int i=1;
138
139	/* fast reuse enable */
140	if (setsockopt(sock,SOL_SOCKET, SO_REUSEADDR, &i, sizeof(i)) < 0)
141	{
142	string err="error setting socket option: ";
143	err+=strerror(errno);
144	log(error, err);
145	throw t2n_server_error(err);
146	}
ac7fdc22	147
0cf4dc9b GE	148	/* keepalive enable */
	149	if (setsockopt(sock,SOL_SOCKET, SO_KEEPALIVE, &i, sizeof(i)) < 0)
	150	{
	151	string err="error setting socket option: ";
	152	err+=strerror(errno);
	153	log(error, err);
	154	throw t2n_server_error(err);
ac7fdc22 GE	155	}
ac7fdc22 GE	156
0cf4dc9b GE	157	/* close on exec */
	158	int fdflags;
	159	fdflags=fcntl(sock,F_GETFD, 0);
	160	if (fdflags < 0)
	161	{
	162	string err="fcntl error on socket: ";
	163	err+=strerror(errno);
	164	log(error, err);
	165	throw t2n_server_error(err);
	166	}
	167	fdflags \|= FD_CLOEXEC;
	168	if (fcntl(sock,F_SETFD,fdflags) < 0)
	169	{
	170	string err="fcntl error on socket: ";
	171	err+=strerror(errno);
	172	log(error, err);
	173	throw t2n_server_error(err);
	174	}
	175
	176	/* non-blocking mode */
	177	int flflags;
	178	flflags=fcntl(sock,F_GETFL,0);
	179	if (flflags < 0)
	180	{
	181	string err="fcntl error on socket: ";
	182	err+=strerror(errno);
	183	log(error, err);
	184	throw t2n_server_error(err);
	185	}
	186	flflags \|= O_NONBLOCK;
	187	if (fcntl(sock,F_SETFL,flflags) < 0)
	188	{
	189	string err="fcntl error on socket: ";
	190	err+=strerror(errno);
	191	log(error, err);
	192	throw t2n_server_error(err);
	193	}
ac7fdc22 GE	194	}
	195
	196	socket_server::~socket_server()
	197	{
0cf4dc9b	198	close(sock);
ac7fdc22	199
0cf4dc9b	200	if (socket_type==unix_s)
ac7fdc22 GE	201	unlink(unix_path.c_str());
	202	}
	203
0cf4dc9b GE	204	void socket_server::new_connection()
0cf4dc9b GE	205	{
04e6b271	206	struct sockaddr_un clientname;
ac7fdc22	207
04e6b271 GE	208	unsigned int size = sizeof (clientname);
	209	int newsock = accept (sock,(struct sockaddr *) &clientname,&size);
	210	if (newsock < 0)
	211	{
	212	if (errno == EAGAIN)
	213	{
	214	log(error, "accept error (EAGAIN): no connection waiting");
	215	return;
	216	}
	217
	218	/* default: break */
	219	string err="error accepting connection: ";
	220	err+=strerror(errno);
	221	log(error, err);
	222	throw t2n_server_error(err);
	223	}
	224
	225	FD_SET (newsock, &connection_set);
	226
	227	int i=1;
	228
	229	/* keepalive enable */
	230	if (setsockopt(newsock,SOL_SOCKET, SO_KEEPALIVE, &i, sizeof(i)) < 0)
	231	{
	232	string err="error setting socket option: ";
	233	err+=strerror(errno);
	234	log(error, err);
	235	throw t2n_server_error(err);
	236	}
	237
	238	/* close on exec */
	239	int fdflags;
	240	fdflags=fcntl(newsock,F_GETFD, 0);
	241	if (fdflags < 0)
	242	{
	243	string err="fcntl error on socket: ";
	244	err+=strerror(errno);
	245	log(error, err);
	246	throw t2n_server_error(err);
	247	}
	248	fdflags \|= FD_CLOEXEC;
	249	if (fcntl(newsock,F_SETFD,fdflags) < 0)
	250	{
	251	string err="fcntl error on socket: ";
	252	err+=strerror(errno);
	253	log(error, err);
	254	throw t2n_server_error(err);
	255	}
	256
	257	/* non-blocking mode */
	258	int flflags;
	259	flflags=fcntl(newsock,F_GETFL,0);
	260	if (flflags < 0)
	261	{
	262	string err="fcntl error on socket: ";
	263	err+=strerror(errno);
	264	log(error, err);
	265	throw t2n_server_error(err);
	266	}
	267	flflags \|= O_NONBLOCK;
	268	if (fcntl(newsock,F_SETFL,flflags) < 0)
	269	{
	270	string err="fcntl error on socket: ";
	271	err+=strerror(errno);
272	log(error, err);
273	throw t2n_server_error(err);
274	}
275
276	add_connection(new socket_connection(newsock, get_default_timeout()));
277
278	return;
0cf4dc9b	279	}
ac7fdc22 GE	280
	281	void socket_server::fill_buffer(long long usec_timeout)
	282	{
0cf4dc9b GE	283	fd_set used_fdset=connection_set;
	284
	285	/* set timeout */
	286	struct timeval tval;
	287	struct timeval *timeout_ptr;
	288
	289	if (usec_timeout == -1)
	290	timeout_ptr = NULL;
	291	else
	292	{
	293	timeout_ptr = &tval;
	294
	295	// timeout von long long usec in int sec + int usec umrechnen
	296	tval.tv_sec = usec_timeout / 1000000;
	297	tval.tv_usec = usec_timeout % 1000000;
	298	}
	299
	300	int ret=select (FD_SETSIZE, &used_fdset, NULL, NULL, timeout_ptr);
	301
	302	if (ret < 0)
	303	{
	304	if (errno == EINTR)
	305	{
	306	// select interrupted by signal
	307	ret=0;
	308	}
	309	else
	310	{
	311	string err="select error: ";
	312	err+=strerror(errno);
	313	log(error, err);
	314	throw t2n_server_error(err);
	315	}
	316	}
	317
	318	if (ret > 0)
	319	{
	320	// we have data pending
	321
	322	// check for new connection
	323	if (FD_ISSET (sock, &used_fdset))
	324	{
	325	new_connection();
	326	}
	327
	328	// check all connections for pending data
	329	fill_connection_buffers();
	330	}
ac7fdc22	331
0cf4dc9b	332	return;
ac7fdc22 GE	333	}
ac7fdc22 GE	334
aa499d20 GE	335	void socket_server::fill_connection_buffers()
	336	{
	337	std::map<unsigned int, connection*>::iterator ie=connections.end();
	338	for(std::map<unsigned int, connection*>::iterator i=connections.begin(); i != ie; i++)
	339	if (!i->second->is_closed())
	340	{
	341	socket_connection* cp=dynamic_cast<socket_connection*>(i->second);
	342	cp->fill_buffer(connection_set);
	343	}
	344	}
	345
04e6b271	346	socket_connection::socket_connection(int _sock, int _timeout)
ac7fdc22 GE	347	: connection(_timeout)
ac7fdc22 GE	348	{
04e6b271	349	sock=_sock;
ac7fdc22 GE	350	}
	351
	352	void socket_connection::close()
	353	{
	354
	355	}
	356
aa499d20	357	void socket_connection::fill_buffer(fd_set &cur_fdset)
ac7fdc22	358	{
aa499d20 GE	359	bool try_again=false;
	360
	361	if (is_closed() \|\| !FD_ISSET (sock, &cur_fdset))
	362	return; // no data pending or connection closed
	363
	364	// data pending -> go and get it
	365	char socket_buffer[recv_buffer_size];
	366
	367	int nbytes = read (sock, socket_buffer, recv_buffer_size);
	368	if (nbytes < 0)
	369	{
	370	if (errno == EAGAIN)
	371	{
	372	if (my_server)
	373	my_server->log(server::error,"read error: no data (EAGAIN) for connection "+get_id_string());
	374	return;
	375	}
	376	else if (errno == EINTR)
	377	{
	378	// interrupted, try again
	379	try_again=true;
	380	}
	381	else
	382	{
	383	if (my_server)
	384	my_server->log(server::error,"error reading from socket of connection "+get_id_string()+": "+strerror(errno));
	385	return;
	386	}
	387	}
	388
	389	// End-of-file
	390	if (nbytes == 0 && !try_again)
	391	{
	392	close();
	393	return;
	394	}
	395
	396	// Data read -> store it
	397	if (nbytes > 0)
	398	buffer.assign(socket_buffer,nbytes);
ac7fdc22	399
aa499d20 GE	400	// more data waiting?
	401	fd_set active_fd_set;
	402	struct timeval tval;
	403
	404	FD_ZERO (&active_fd_set);
	405	FD_SET (sock, &active_fd_set);
	406
	407	/* no waiting */
	408	tval.tv_sec=0;
	409	tval.tv_usec=0;
	410
	411	if (select (FD_SETSIZE, &active_fd_set, NULL, NULL, &tval) >0)
	412	{
	413	/* more data waiting -> recurse */
	414	fill_buffer(active_fd_set);
	415	}
	416
	417	reset_timeout();
	418
	419	return;
ac7fdc22 GE	420	}
	421
	422	void socket_connection::write(const std::string& data)
	423	{
aa499d20 GE	424	static const unsigned int write_block_size=4096;
	425
	426	if (is_closed())
	427	return;
	428
	429	// prepend packet size to data
	430	packet_size_indicator psize=data.size();
	431	string send_data(data);
	432	send_data.insert(0,(char*)psize,sizeof(packet_size_indicator));
	433
	434	int offset = 0;
	435	while (offset < send_data.size())
	436	{
	437	unsigned int write_size=write_block_size;
	438
	439	if (offset+write_size > send_data.size())
	440	write_size = send_data.size()-offset;
ac7fdc22	441
aa499d20 GE	442	int rtn;
	443	while ((rtn=::write(sock, send_data.data()+offset, write_size)) &&
	444	rtn == -1 && (errno == EAGAIN \|\| errno == EINTR))
	445	{
	446	usleep (80000);
	447	if (my_server)
	448	my_server->log(server::debug,"resuming write() call after EAGAIN or EINTR for connection "+get_id_string());
	449	}
	450
	451	if (rtn == -1)
	452	{
	453	if (my_server)
	454	my_server->log(server::error,"write() error on connection "+get_id_string()+": "+strerror(errno));
	455	}
	456	else if (rtn != write_size)
	457	{
	458	if (my_server)
	459	{
	460	ostringstream msg;
	461	msg << "write() error on connection " << get_id()
	462	<< ": wrote " << rtn << " bytes, should have been "
	463	<< write_size << " (complete: " << send_data.size() << ")";
	464
	465	my_server->log(server::error,msg.str());
	466	}
	467	}
	468
	469	offset += write_size;
	470	}
	471
	472	reset_timeout();
	473
	474	return;
ac7fdc22 GE	475	}
	476
	477	}