path: root/wrapper/ppswrapper-common.c

/* PPSWrapper

 * Copyright (C) 2009- Luo Jinghua <sunmoon1997@gmail.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * 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
 * Library General Public License for more
 */
#include <assert.h>
#include <stdlib.h>
#include <stdio.h>
#include <ctype.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>

#include "pps-fixed.h"
#include "ppswrapper.h"

static ssize_t
safe_read(int fildes, void *buf, size_t nbytes)
{
    ssize_t ret;
    int error;
    size_t left = nbytes;

    do {
	ret = read(fildes, (char*)buf + nbytes - left, left);
	if (ret > 0 && ret < left) {
	    //printf ("partial read: %zd left %zd\n", ret, left);
	    left -= ret;
	    ret = -1;
	    errno = EINTR;
	} else if (ret == left) {
	    left = 0;
	    break;
	} else if (ret == 0) {
	    error = errno;
	    //printf ("read zero bytes left: %zd\n", left);
	    errno = error;
	}
    } while (ret < 0 && (errno == EINTR || errno == EAGAIN));

    if (ret < 0) {
	error = errno;
	fprintf (stderr, "Failed to read: errno %d(%s)\n", error, strerror(errno));
	errno = error;
    }

    if (ret < 0)
	return ret;
    return nbytes - left;
}

static ssize_t
safe_write(int fildes, const void *buf, size_t nbytes)
{
    ssize_t ret;
    size_t left = nbytes;
    int error;

    do {
	ret = write(fildes, (const char*)buf + nbytes - left, left);
	if (ret > 0 && ret < left) {
	    //printf ("partial write: %zd left %zd\n", ret, left);
	    left -= ret;
	    ret = -1;
	    errno = EINTR;
	} else if (ret == left) {
	    left = 0;
	    break;
	} else if (ret == 0) {
	    error = errno;
	    //printf ("write zero bytes left: %zd\n", left);
	    errno = error;
	}
    } while (ret < 0 && (errno == EINTR || errno == EAGAIN || errno == EINPROGRESS));

    if (ret < 0) {
	error = errno;
	fprintf (stderr, "Failed to write: errno: %d(%s)\n", error, strerror(errno));
	errno = error;
    }

    if (ret < 0)
	return ret;
    return nbytes - left;
}

int
ppsw_read(ppsw_t* self,
	  uint8_t* retval,
	  size_t len)
{
    int ret;

    ret = safe_read(self->sock, retval, len);
    if (ret != len)
	return -1;

    return ret;
}

int
ppsw_write(ppsw_t* self,
	   const uint8_t* retval,
	   size_t len)
{
    int ret;

    ret = safe_write(self->sock, retval, len);
    if (ret != len)
	return -1;
    return ret;
}

int
ppsw_readl(ppsw_t* self,
	   uint32_t* retval)
{
    uint8_t val[4];
    int ret;

    ret = ppsw_read(self, val, 4);
    if (ret != 4)
	return -1;

    *retval =
	val[0] << 24 | val[1] << 16 |
	val[2] << 8 | val[3];

    return ret;
}

int
ppsw_writel(ppsw_t* self,
	   uint32_t retval)
{
    uint8_t val[4];
    int ret;

    val[0] = (retval >> 24) & 0xff;
    val[1] = (retval >> 16) & 0xff;
    val[2] = (retval >>  8) & 0xff;
    val[3] = (retval >>  0) & 0xff;

    ret = ppsw_write(self, val, 4);
    if (ret != 4)
	return -1;
    return ret;
}

int
ppsw_reads(ppsw_t* self,
	   uint16_t* retval)
{
    uint8_t val[2];
    int ret;

    ret = ppsw_read(self, val, 2);
    if (ret != 2)
	return -1;

    *retval = val[0] << 8 | val[1];

    return ret;
}

int
ppsw_writes(ppsw_t* self,
	    uint16_t retval)
{
    uint8_t val[2];
    int ret;

    val[0] = (retval >>  8) & 0xff;
    val[1] = (retval >>  0) & 0xff;

    ret = ppsw_write(self, val, 2);
    if (ret != 2)
	return -1;
    return ret;
}

int
ppsw_readb(ppsw_t* self,
	   uint8_t* retval)
{
    uint8_t val;
    int ret;

    ret = ppsw_read(self, &val, 1);
    if (ret != 1) {
	perror("Failed to read a byte from peer");
	return -1;
    }

    *retval = val;
    return ret;
}

int
ppsw_writeb(ppsw_t* self,
	    uint8_t retval)
{
    int ret;

    ret = ppsw_write(self, &retval, 1);
    if (ret != 1)
	return -1;
    return ret;
}

int
ppsw_read_str(ppsw_t* self,
	      char** retval,
	      uint16_t* retlen)
{
    int ret;
    uint16_t len;
    char* val;

    ret = ppsw_reads(self, &len);
    if (ret < 0)
	return -1;

    if (len == 0) {
	*retval = NULL;
	return 2;
    }

    val = malloc(len);
    if (!val)
	return -1;

    ret = ppsw_read(self, (uint8_t*)val, len);
    if (ret != len) {
	free (val);
	return -1;
    }

    *retval = val;
    if (retlen)
	*retlen = len;
    return 2 + len;
}

int
ppsw_write_str(ppsw_t* self,
	       const char* retval,
	       int len)
{
    int ret;

    if (len < 0)
	len = retval ? strlen (retval) + 1 : 0;
    if (ppsw_writes (self, len) != 2)
	return -1;
    if (!len)
	return 2;
    ret = ppsw_write (self, (const uint8_t*)retval, len);
    if (ret != len)
	return -1;
    return 2 + len;
}


int
ppsw_read_msg(ppsw_t* self,
	      ppsw_msg_t* msg)
{
    if (ppsw_readb(self, &msg->sync) < 0) {
	perror("Failed to read message.sync from peer");
	return -1;
    }
    if (msg->sync != PPSW_SYNC) {
	fprintf(stderr, "%d unexpect sync: %d exepected: %d\n",
		getpid(), msg->sync, PPSW_SYNC);
	return -1;
    }
    if (ppsw_reads(self, &msg->id) < 0) {
	perror("Failed to read message.id from peer");
	return -1;
    }
    if (ppsw_reads(self, &msg->seq) < 0) {
	perror("Failed to read message.seqno from peer");
	return -1;
    }
    if (ppsw_readl(self, &msg->len) < 0) {
	perror("Failed to read message.length from peer");
	return -1;
    }

    return 0;
}

int
ppsw_read_reply_msg(ppsw_t* self,
		    ppsw_reply_msg_t* msg)
{
    ppsw_packet_t p;

    if (ppsw_recv_packet(self, &msg->base, &p) < 0)
	return -1;
    if (ppsw_packet_readl(&p, &msg->retval) < 0)
	return -1;
    ppsw_packet_fini(&p);
    return 0;
}

int
ppsw_send_reply_msg(ppsw_t* self,
		    ppsw_reply_msg_t* msg)
{
    ppsw_packet_t p;

    if (ppsw_packet_init(&p, msg->base.id, 4) < 0)
	return -1;
    if (ppsw_packet_writel(&p, msg->retval) < 0)
	return -1;
    if (ppsw_reply_packet(self, msg->base.seq, &p) < 0)
	return -1;
    return 0;
}

size_t
ppsw_strlen(const char* s)
{
    if (!s)
	return 0;
    return strlen(s) + 1;
}

int
ppsw_write_item_info(ppsw_packet_t* self,
		     ppsvod_iteminfo* info)
{
    if (ppsw_packet_writel(self, info->index) < 0)
	return -1;
    if (ppsw_packet_write_str(self, info->hash_id, HASH_VALUE_LENGTH) < 0)
	return -1;
    if (ppsw_packet_write_str(self, info->file_name, -1) < 0)
	return -1;
    if (ppsw_packet_writel(self, info->file_size) < 0)
	return -1;
    if (ppsw_packet_writel(self, info->progress) < 0)
	return -1;
    if (ppsw_packet_writel(self, info->down_speed) < 0)
	return -1;
    if (ppsw_packet_writel(self, info->up_speed) < 0)
	return -1;
    if (ppsw_packet_writel(self, info->peer_count) < 0)
	return -1;
    if (ppsw_packet_writel(self, info->status) < 0)
	return -1;
    return 0;
}

int
ppsw_read_item_info(ppsw_packet_t* self,
		    ppsvod_iteminfo* info)
{
    char *hash_id;
    uint32_t val;

    if (ppsw_packet_readl(self, &val) < 0)
	return -1;
    info->index = val;

    if (ppsw_packet_read_str(self, &hash_id, NULL) < 0)
	return -1;
    memcpy(info->hash_id, hash_id, sizeof(info->hash_id));
    free (hash_id);

    if (ppsw_packet_read_str(self, &info->file_name, NULL) < 0)
	goto failed;

    if (ppsw_packet_readl(self, &val) < 0)
	goto failed;
    info->file_size = val;

    if (ppsw_packet_readl(self, &val) < 0)
	goto failed;
    info->progress = val;

    if (ppsw_packet_readl(self, &val) < 0)
	goto failed;
    info->down_speed = val;

    if (ppsw_packet_readl(self, &val) < 0)
	goto failed;
    info->up_speed = val;

    if (ppsw_packet_readl(self, &val) < 0)
	goto failed;
    info->peer_count = val;

    if (ppsw_packet_readl(self, &val) < 0)
	goto failed;
    info->status = val;
    return 0;
 failed:
    free (info->file_name);
    return -1;
}

int
ppsw_packet_readl(ppsw_packet_t* self,
		  uint32_t* retval)
{
    uint8_t val[4];
    int ret;

    ret = ppsw_packet_read(self, val, 4);
    if (ret != 4) {
	ppsw_packet_fini(self);
	return -1;
    }

    *retval =
	val[0] << 24 | val[1] << 16 |
	val[2] << 8 | val[3];

    return 4;
}

int
ppsw_packet_writel(ppsw_packet_t* self,
		   uint32_t retval)
{
    uint8_t val[4];
    int ret;

    val[0] = (retval >> 24) & 0xff;
    val[1] = (retval >> 16) & 0xff;
    val[2] = (retval >>  8) & 0xff;
    val[3] = (retval >>  0) & 0xff;

    ret = ppsw_packet_write(self, val, 4);
    if (ret < 0) {
	ppsw_packet_fini(self);
	return -1;
    }
    return ret;
}

int
ppsw_packet_reads(ppsw_packet_t* self,
		  uint16_t* retval)
{
    uint8_t val[2];
    int ret;

    ret = ppsw_packet_read(self, val, 2);
    if (ret != 2) {
	ppsw_packet_fini(self);
	return -1;
    }

    *retval = val[0] << 8 | val[1];
    return 2;
}

int
ppsw_packet_writes(ppsw_packet_t* self,
		   uint16_t retval)
{
    uint8_t val[2];
    int ret;

    val[0] = (retval >>  8) & 0xff;
    val[1] = (retval >>  0) & 0xff;

    ret = ppsw_packet_write(self, val, 2);
    if (ret < 0) {
	ppsw_packet_fini(self);
	return -1;
    }
    return 2;
}

int
ppsw_packet_readb(ppsw_packet_t* self,
		  uint8_t* retval)
{
    uint8_t val;
    int ret;

    ret = ppsw_packet_read(self, &val, 1);
    if (ret != 1) {
	perror("Failed to read a byte from peer");
	ppsw_packet_fini(self);
	errno = ENOSPC;
	return -1;
    }

    *retval = val;
    return ret;
}

int
ppsw_packet_writeb(ppsw_packet_t* self,
		   uint8_t retval)
{
    int ret;

    ret = ppsw_packet_write(self, &retval, 1);
    if (ret < 0) {
	ppsw_packet_fini(self);
	return -1;
    }
    return 1;
}

int
ppsw_packet_read(ppsw_packet_t* self,
		 uint8_t* retval,
		 uint32_t len)
{
    if (self->len - self->offset < len) {
	ppsw_packet_fini(self);
	errno = ENOSPC;
	return -1;
    }

    memcpy(retval, &self->data[self->offset], len);
    self->offset += len;
    return len;
}

int
ppsw_packet_write(ppsw_packet_t* self,
		  const uint8_t* val,
		  uint32_t len)
{
    if (self->len - self->offset < len) {
	ppsw_packet_fini(self);
	return -1;
    }

    memcpy(&self->data[self->offset], val, len);
    self->offset += len;
    return len;
}

int
ppsw_packet_read_str(ppsw_packet_t* self,
		     char** retval,
		     uint16_t* retlen)
{
    int ret;
    uint16_t len;
    char* val;

    ret = ppsw_packet_reads(self, &len);
    if (ret < 0)
	return -1;

    if (len == 0) {
	*retval = NULL;
	return 2;
    }

    val = malloc(len);
    if (!val) {
	ppsw_packet_fini(self);
	return -1;
    }

    ret = ppsw_packet_read(self, (uint8_t*)val, len);
    if (ret != len) {
	free (val);
	return -1;
    }

    *retval = val;
    if (retlen)
	*retlen = len;
    return 2 + len;
}

int
ppsw_packet_write_str(ppsw_packet_t* self,
		      const char* retval,
		      int len)
{
    int ret;

    if (len < 0)
	len = retval ? strlen (retval) + 1 : 0;
    assert (len < 0xffff);
    if (ppsw_packet_writes (self, len) < 0)
	return -1;
    if (!len)
	return 2;
    ret = ppsw_packet_write (self, (const uint8_t*)retval, len);
    if (ret < 0)
	return -1;
    return 2 + len;
}

int
ppsw_packet_init(ppsw_packet_t* p,
		 uint16_t id,
		 uint32_t len)
{
    if (len > 0xffffff)
	return -1;

    p->offset = 0;
    p->len = PPSW_MSG_HEADER_LEN + len;
    if (PPSW_MSG_HEADER_LEN + len < 64)
	p->data = p->local;
    else
	p->data = malloc(PPSW_MSG_HEADER_LEN + len);
    if (!p->data)
	return -1;

    ppsw_packet_writeb(p, PPSW_SYNC);
    ppsw_packet_writes(p, id);
    /* skip seqno */
    p->offset += 2;
    ppsw_packet_writel(p, len);
    return 0;
}

void
ppsw_packet_fini(ppsw_packet_t* p)
{
    if (p->data != p->local)
	free (p->data);
    p->data = NULL;
}

int
ppsw_packet_reinit(ppsw_packet_t* p,
		   uint16_t id,
		   uint32_t len)
{
    ppsw_packet_fini(p);
    return ppsw_packet_init(p, id, len);
}

int
ppsw_send_packet(ppsw_t* self,
		 ppsw_msg_t* msg,
		 ppsw_packet_t* p)
{
    int ret;

    pthread_mutex_lock (&self->packet_mutex);

    p->offset = 1 + 2;
    if (msg)
	msg->seq = self->seq;
    ppsw_packet_writes(p, self->seq++);

    ret = ppsw_write(self, p->data, p->len);

#if 0
    fprintf (stderr, "%d: send packet: id %d seqno %d -> %d\n",
	     getpid(), p->data[1] << 8 | p->data[2], self->seq - 1, self->seq);
#endif

    pthread_mutex_unlock (&self->packet_mutex);

    ppsw_packet_fini(p);

    return ret;
}

int
ppsw_reply_packet(ppsw_t* self,
		  uint16_t seq,
		  ppsw_packet_t* p)
{
    int ret;

    pthread_mutex_lock (&self->packet_mutex);

    p->offset = 1 + 2;
    ppsw_packet_writes(p, seq);

    ret = ppsw_write(self, p->data, p->len);

#if 0
    fprintf (stderr, "%d: reply packet: id %d seqno %d\n",
	     getpid(), p->data[1] << 8 | p->data[2], seq);
#endif

    pthread_mutex_unlock (&self->packet_mutex);

    ppsw_packet_fini(p);

    return ret;
}

int
ppsw_recv_packet(ppsw_t* self,
		 ppsw_msg_t* msg,
		 ppsw_packet_t* p)
{
    int ret;

    pthread_mutex_lock (&self->packet_mutex);


    if (ppsw_read_msg(self, msg) < 0) {
	ret = -1;
	goto done;
    }

    if (ppsw_packet_init(p, msg->id, msg->len) < 0) {
	int i;
	uint8_t val;

	perror("Failed to initialize packet");
	for (i = 0; i < msg->len; i++)
	    if (ppsw_read(self, &val, 1) < 0)
		break;

	ret = -1;
	goto done;
    }

    p->offset = 1 + 2;
    ppsw_packet_writes(p, msg->seq);
    p->offset = PPSW_MSG_HEADER_LEN;

    ret = ppsw_read(self, &p->data[p->offset], msg->len);
    if (ret < 0) {
	perror("Failed to recv packet content");
	//__asm__ ("int $3");
	ppsw_packet_fini(p);
	ret = -1;
	goto done;
    }

#if 0
    fprintf (stderr, "%d: recv packet: id %d seqno %d\n",
	     getpid(), msg->id, msg->seq);
#endif

 done:
    pthread_mutex_unlock (&self->packet_mutex);

    return ret;
}

void
ppsw_init(ppsw_t* self)
{
    pthread_mutex_init (&self->packet_mutex, NULL);
    self->seq = 0;
}

void
ppsw_fini(ppsw_t* self)
{
    pthread_mutex_destroy (&self->packet_mutex);
}