Browse Source

Btpd now uses evloop, it's own event loop, instead of libevent.

master
Richard Nyberg 16 years ago
parent
commit
eb421cc586
15 changed files with 160 additions and 125 deletions
  1. +27
    -20
      btpd/btpd.c
  2. +8
    -3
      btpd/btpd.h
  3. +5
    -8
      btpd/cli_if.c
  4. +4
    -4
      btpd/content.c
  5. +19
    -21
      btpd/http_tr_if.c
  6. +3
    -10
      btpd/main.c
  7. +8
    -7
      btpd/nameconn.c
  8. +30
    -17
      btpd/net.c
  9. +1
    -2
      btpd/net.h
  10. +1
    -2
      btpd/net_types.h
  11. +4
    -7
      btpd/peer.c
  12. +2
    -3
      btpd/thread_cb.c
  13. +12
    -12
      btpd/tracker_req.c
  14. +5
    -5
      btpd/upload.c
  15. +31
    -4
      btpd/util.c

+ 27
- 20
btpd/btpd.c View File

@@ -4,9 +4,9 @@
#include <signal.h> #include <signal.h>


static uint8_t m_peer_id[20]; static uint8_t m_peer_id[20];
static struct event m_sigint;
static struct event m_sigterm;
static struct event m_heartbeat;
static struct timeout m_heartbeat;
static struct timeout m_grace_timer;
static int m_signal;
static int m_shutdown; static int m_shutdown;


long btpd_seconds; long btpd_seconds;
@@ -37,11 +37,9 @@ btpd_shutdown(int grace_seconds)
BTPDQ_FOREACH(tp, torrent_get_all(), entry) BTPDQ_FOREACH(tp, torrent_get_all(), entry)
if (tp->state != T_STOPPING) if (tp->state != T_STOPPING)
torrent_stop(tp, 0); torrent_stop(tp, 0);
if (grace_seconds >= 0) {
if (event_once(-1, EV_TIMEOUT, grace_cb, NULL,
(& (struct timeval) { grace_seconds, 0 })) != 0)
btpd_err("failed to add event (%s).\n", strerror(errno));
}
if (grace_seconds >= 0)
btpd_timer_add(&m_grace_timer,
(& (struct timespec){ grace_seconds, 0 }));
} }
} }


@@ -57,19 +55,23 @@ btpd_get_peer_id(void)
} }


static void static void
signal_cb(int signal, short type, void *arg)
signal_handler(int signal)
{ {
btpd_log(BTPD_L_BTPD, "Got signal %d.\n", signal);
btpd_shutdown(30);
m_signal = signal;
} }


static void static void
heartbeat_cb(int fd, short type, void *arg) heartbeat_cb(int fd, short type, void *arg)
{ {
btpd_ev_add(&m_heartbeat, (& (struct timeval) { 1, 0 }));
btpd_timer_add(&m_heartbeat, (& (struct timespec) { 1, 0 }));
btpd_seconds++; btpd_seconds++;
net_on_tick(); net_on_tick();
torrent_on_tick_all(); torrent_on_tick_all();
if (m_signal) {
btpd_log(BTPD_L_BTPD, "Got signal %d.\n", m_signal);
btpd_shutdown(30);
m_signal = 0;
}
if (m_shutdown && torrent_count() == 0) if (m_shutdown && torrent_count() == 0)
btpd_exit(0); btpd_exit(0);
} }
@@ -82,11 +84,21 @@ void addrinfo_init(void);
void void
btpd_init(void) btpd_init(void)
{ {
struct sigaction sa;
unsigned long seed; unsigned long seed;
uint8_t idcon[1024]; uint8_t idcon[1024];
struct timeval now; struct timeval now;
int n; int n;


bzero(&sa, sizeof(sa));
sa.sa_handler = SIG_IGN;
sa.sa_flags = SA_RESTART;
sigfillset(&sa.sa_mask);
sigaction(SIGPIPE, &sa, NULL);
sa.sa_handler = signal_handler;
sigaction(SIGTERM, &sa, NULL);
sigaction(SIGINT, &sa, NULL);

gettimeofday(&now, NULL); gettimeofday(&now, NULL);
n = snprintf(idcon, sizeof(idcon), "%ld%ld%d", now.tv_sec, now.tv_usec, n = snprintf(idcon, sizeof(idcon), "%ld%ld%d", now.tv_sec, now.tv_usec,
net_port); net_port);
@@ -111,14 +123,9 @@ btpd_init(void)
tr_init(); tr_init();
tlib_init(); tlib_init();


signal(SIGPIPE, SIG_IGN);

signal_set(&m_sigint, SIGINT, signal_cb, NULL);
btpd_ev_add(&m_sigint, NULL);
signal_set(&m_sigterm, SIGTERM, signal_cb, NULL);
btpd_ev_add(&m_sigterm, NULL);
evtimer_set(&m_heartbeat, heartbeat_cb, NULL);
btpd_ev_add(&m_heartbeat, (& (struct timeval) { 1, 0 }));
timer_init(&m_grace_timer, grace_cb, NULL);
timer_init(&m_heartbeat, heartbeat_cb, NULL);
btpd_timer_add(&m_heartbeat, (& (struct timespec) { 1, 0 }));


if (!empty_start) if (!empty_start)
active_start(); active_start();


+ 8
- 3
btpd/btpd.h View File

@@ -26,7 +26,7 @@
#define DAEMON #define DAEMON
#include <btpd_if.h> #include <btpd_if.h>
#undef DAEMON #undef DAEMON
#include <event.h>
#include <evloop.h>
#include <metainfo.h> #include <metainfo.h>
#include <queue.h> #include <queue.h>
#include <subr.h> #include <subr.h>
@@ -70,8 +70,13 @@ void *btpd_malloc(size_t size);
__attribute__((malloc)) __attribute__((malloc))
void *btpd_calloc(size_t nmemb, size_t size); void *btpd_calloc(size_t nmemb, size_t size);


void btpd_ev_add(struct event *ev, struct timeval *tv);
void btpd_ev_del(struct event *ev);
void btpd_ev_new(struct fdev *ev, int fd, uint16_t flags, evloop_cb_t cb,
void *arg);
void btpd_ev_del(struct fdev *ev);
void btpd_ev_enable(struct fdev *ev, uint16_t flags);
void btpd_ev_disable(struct fdev *ev, uint16_t flags);
void btpd_timer_add(struct timeout *to, struct timespec *ts);
void btpd_timer_del(struct timeout *to);


void btpd_shutdown(int grace_seconds); void btpd_shutdown(int grace_seconds);
int btpd_is_stopping(void); int btpd_is_stopping(void);


+ 5
- 8
btpd/cli_if.c View File

@@ -5,10 +5,10 @@


struct cli { struct cli {
int sd; int sd;
struct event read;
struct fdev read;
}; };


static struct event m_cli_incoming;
static struct fdev m_cli_incoming;


static int static int
write_buffer(struct cli *cli, struct iobuf *iob) write_buffer(struct cli *cli, struct iobuf *iob)
@@ -420,10 +420,10 @@ cli_read_cb(int sd, short type, void *arg)
goto error; goto error;


free(msg); free(msg);
btpd_ev_add(&cli->read, NULL);
return; return;


error: error:
btpd_ev_del(&cli->read);
close(cli->sd); close(cli->sd);
free(cli); free(cli);
if (msg != NULL) if (msg != NULL)
@@ -447,8 +447,7 @@ client_connection_cb(int sd, short type, void *arg)


struct cli *cli = btpd_calloc(1, sizeof(*cli)); struct cli *cli = btpd_calloc(1, sizeof(*cli));
cli->sd = nsd; cli->sd = nsd;
event_set(&cli->read, cli->sd, EV_READ, cli_read_cb, cli);
btpd_ev_add(&cli->read, NULL);
btpd_ev_new(&cli->read, cli->sd, EV_READ, cli_read_cb, cli);
} }


void void
@@ -478,7 +477,5 @@ ipc_init(void)
listen(sd, 4); listen(sd, 4);
set_nonblocking(sd); set_nonblocking(sd);


event_set(&m_cli_incoming, sd, EV_READ | EV_PERSIST,
client_connection_cb, NULL);
btpd_ev_add(&m_cli_incoming, NULL);
btpd_ev_new(&m_cli_incoming, sd, EV_READ, client_connection_cb, NULL);
} }

+ 4
- 4
btpd/content.c View File

@@ -53,7 +53,7 @@ BTPDQ_HEAD(std_tq, start_test_data);


static struct std_tq m_startq = BTPDQ_HEAD_INITIALIZER(m_startq); static struct std_tq m_startq = BTPDQ_HEAD_INITIALIZER(m_startq);


static struct event m_workev;
static struct timeout m_workev;


#define READBUFLEN (1 << 14) #define READBUFLEN (1 << 14)


@@ -456,7 +456,7 @@ startup_test_run(void)
if (std->start >= tp->npieces) if (std->start >= tp->npieces)
startup_test_end(tp, 1); startup_test_end(tp, 1);
if (!BTPDQ_EMPTY(&m_startq)) if (!BTPDQ_EMPTY(&m_startq))
event_add(&m_workev, (& (struct timeval) { 0, 0 }));
btpd_timer_add(&m_workev, (& (struct timespec) { 0, 0 }));
} }


void void
@@ -473,7 +473,7 @@ startup_test_begin(struct torrent *tp, struct file_time_size *fts)
std->fts = fts; std->fts = fts;
BTPDQ_INSERT_TAIL(&m_startq, std, entry); BTPDQ_INSERT_TAIL(&m_startq, std, entry);
if (std == BTPDQ_FIRST(&m_startq)) if (std == BTPDQ_FIRST(&m_startq))
event_add(&m_workev, (& (struct timeval) { 0, 0 }));
btpd_timer_add(&m_workev, (& (struct timespec) { 0, 0 }));
} else { } else {
free(fts); free(fts);
startup_test_end(tp, 0); startup_test_end(tp, 0);
@@ -539,5 +539,5 @@ cm_start(struct torrent *tp, int force_test)
void void
cm_init(void) cm_init(void)
{ {
evtimer_set(&m_workev, worker_cb, NULL);
timer_init(&m_workev, worker_cb, NULL);
} }

+ 19
- 21
btpd/http_tr_if.c View File

@@ -11,8 +11,7 @@ struct http_tr_req {
struct torrent *tp; struct torrent *tp;
struct http_req *req; struct http_req *req;
struct iobuf buf; struct iobuf buf;
struct event rdev;
struct event wrev;
struct fdev ioev;
nameconn_t nc; nameconn_t nc;
int sd; int sd;
enum tr_event event; enum tr_event event;
@@ -22,8 +21,7 @@ static void
http_tr_free(struct http_tr_req *treq) http_tr_free(struct http_tr_req *treq)
{ {
if (treq->sd != -1) { if (treq->sd != -1) {
btpd_ev_del(&treq->rdev);
btpd_ev_del(&treq->wrev);
btpd_ev_del(&treq->ioev);
close(treq->sd); close(treq->sd);
} }
iobuf_free(&treq->buf); iobuf_free(&treq->buf);
@@ -143,19 +141,21 @@ http_cb(struct http_req *req, struct http_response *res, void *arg)
} }


static void static void
sd_wr_cb(int sd, short type, void *arg)
sd_io_cb(int sd, short type, void *arg)
{ {
struct http_tr_req *treq = arg; struct http_tr_req *treq = arg;
if (http_write(treq->req, sd) && http_want_write(treq->req))
btpd_ev_add(&treq->wrev, NULL);
}

static void
sd_rd_cb(int sd, short type, void *arg)
{
struct http_tr_req *treq = arg;
if (http_read(treq->req, sd) && http_want_read(treq->req))
btpd_ev_add(&treq->rdev, NULL);
switch (type) {
case EV_READ:
if (http_read(treq->req, sd) && !http_want_read(treq->req))
btpd_ev_disable(&treq->ioev, EV_READ);
break;
case EV_WRITE:
if (http_write(treq->req, sd) && !http_want_write(treq->req))
btpd_ev_disable(&treq->ioev, EV_WRITE);
break;
default:
abort();
}
} }


static void static void
@@ -168,12 +168,10 @@ nc_cb(void *arg, int error, int sd)
http_tr_free(treq); http_tr_free(treq);
} else { } else {
treq->sd = sd; treq->sd = sd;
event_set(&treq->wrev, sd, EV_WRITE, sd_wr_cb, treq);
event_set(&treq->rdev, sd, EV_READ, sd_rd_cb, treq);
if (http_want_read(treq->req))
btpd_ev_add(&treq->rdev, NULL);
if (http_want_write(treq->req))
btpd_ev_add(&treq->wrev, NULL);
uint16_t flags =
(http_want_read(treq->req) ? EV_READ : 0) |
(http_want_write(treq->req) ? EV_WRITE : 0);
btpd_ev_new(&treq->ioev, sd, flags, sd_io_cb, treq);
} }
} }




+ 3
- 10
btpd/main.c View File

@@ -2,7 +2,6 @@


#include <sys/file.h> #include <sys/file.h>
#include <err.h> #include <err.h>
#include <evdns.h>
#include <getopt.h> #include <getopt.h>


static void static void
@@ -222,19 +221,13 @@ args_done:


setup_daemon(daemonize, dir, log); setup_daemon(daemonize, dir, log);


event_init();

if ((errno = evdns_resolv_conf_parse(DNS_OPTION_NAMESERVERS,
"/etc/resolv.conf")) != 0) {
btpd_log(BTPD_L_ERROR,
"failed to setup dns from /etc/resolv.conf (%d).\n", errno);
}
evloop_init();


btpd_init(); btpd_init();


event_dispatch();
evloop();


btpd_err("Unexpected exit from libevent.\n");
btpd_err("Unexpected exit from evloop.\n");


return 1; return 1;
} }

+ 8
- 7
btpd/nameconn.c View File

@@ -1,7 +1,7 @@
#include "btpd.h" #include "btpd.h"


struct nameconn { struct nameconn {
struct event write_ev;
struct fdev write_ev;
void (*cb)(void *, int, int); void (*cb)(void *, int, int);
void *arg; void *arg;
aictx_t ai_handle; aictx_t ai_handle;
@@ -35,9 +35,11 @@ nc_write_cb(int sd, short type, void *arg)
socklen_t errsiz = sizeof(int); socklen_t errsiz = sizeof(int);
if (getsockopt(nc->sd, SOL_SOCKET, SO_ERROR, &error, &errsiz) < 0) if (getsockopt(nc->sd, SOL_SOCKET, SO_ERROR, &error, &errsiz) < 0)
btpd_err("getsockopt error (%s).\n", strerror(errno)); btpd_err("getsockopt error (%s).\n", strerror(errno));
if (error == 0)
if (error == 0) {
btpd_ev_del(&nc->write_ev);
nc_done(nc, 0); nc_done(nc, 0);
else {
} else {
btpd_ev_del(&nc->write_ev);
close(nc->sd); close(nc->sd);
nc->ai_cur = nc->ai_cur->ai_next; nc->ai_cur = nc->ai_cur->ai_next;
nc_connect(nc); nc_connect(nc);
@@ -64,10 +66,9 @@ again:
err = connect(nc->sd, ai->ai_addr, ai->ai_addrlen); err = connect(nc->sd, ai->ai_addr, ai->ai_addrlen);
if (err == 0) if (err == 0)
nc_done(nc, 0); nc_done(nc, 0);
else if (err < 0 && errno == EINPROGRESS) {
event_set(&nc->write_ev, nc->sd, EV_WRITE, nc_write_cb, nc);
btpd_ev_add(&nc->write_ev, NULL);
} else {
else if (err < 0 && errno == EINPROGRESS)
btpd_ev_new(&nc->write_ev, nc->sd, EV_WRITE, nc_write_cb, nc);
else {
close(nc->sd); close(nc->sd);
nc->ai_cur = ai->ai_next; nc->ai_cur = ai->ai_next;
goto again; goto again;


+ 30
- 17
btpd/net.c View File

@@ -9,7 +9,7 @@ static unsigned long m_bw_bytes_out;
static unsigned long m_rate_up; static unsigned long m_rate_up;
static unsigned long m_rate_dwn; static unsigned long m_rate_dwn;


static struct event m_net_incoming;
static struct fdev m_net_incoming;


unsigned net_npeers; unsigned net_npeers;


@@ -154,7 +154,6 @@ net_write(struct peer *p, unsigned long wmax)
nwritten = writev(p->sd, iov, niov); nwritten = writev(p->sd, iov, niov);
if (nwritten < 0) { if (nwritten < 0) {
if (errno == EAGAIN) { if (errno == EAGAIN) {
btpd_ev_add(&p->out_ev, NULL);
p->t_wantwrite = btpd_seconds; p->t_wantwrite = btpd_seconds;
return 0; return 0;
} else { } else {
@@ -194,10 +193,10 @@ net_write(struct peer *p, unsigned long wmax)
bcount = 0; bcount = 0;
} }
} }
if (!BTPDQ_EMPTY(&p->outq)) {
btpd_ev_add(&p->out_ev, NULL);
if (!BTPDQ_EMPTY(&p->outq))
p->t_wantwrite = btpd_seconds; p->t_wantwrite = btpd_seconds;
}
else
btpd_ev_disable(&p->ioev, EV_WRITE);
p->t_lastwrite = btpd_seconds; p->t_lastwrite = btpd_seconds;


return nwritten; return nwritten;
@@ -435,7 +434,6 @@ net_read(struct peer *p, unsigned long rmax)
} }


out: out:
btpd_ev_add(&p->in_ev, NULL);
return nread > 0 ? nread : 0; return nread > 0 ? nread : 0;
} }


@@ -557,12 +555,14 @@ net_bw_tick(void)
if (net_bw_limit_in > 0) { if (net_bw_limit_in > 0) {
while ((p = BTPDQ_FIRST(&net_bw_readq)) != NULL && m_bw_bytes_in > 0) { while ((p = BTPDQ_FIRST(&net_bw_readq)) != NULL && m_bw_bytes_in > 0) {
BTPDQ_REMOVE(&net_bw_readq, p, rq_entry); BTPDQ_REMOVE(&net_bw_readq, p, rq_entry);
btpd_ev_enable(&p->ioev, EV_READ);
p->flags &= ~PF_ON_READQ; p->flags &= ~PF_ON_READQ;
m_bw_bytes_in -= net_read(p, m_bw_bytes_in); m_bw_bytes_in -= net_read(p, m_bw_bytes_in);
} }
} else { } else {
while ((p = BTPDQ_FIRST(&net_bw_readq)) != NULL) { while ((p = BTPDQ_FIRST(&net_bw_readq)) != NULL) {
BTPDQ_REMOVE(&net_bw_readq, p, rq_entry); BTPDQ_REMOVE(&net_bw_readq, p, rq_entry);
btpd_ev_enable(&p->ioev, EV_READ);
p->flags &= ~PF_ON_READQ; p->flags &= ~PF_ON_READQ;
net_read(p, 0); net_read(p, 0);
} }
@@ -572,12 +572,14 @@ net_bw_tick(void)
while (((p = BTPDQ_FIRST(&net_bw_writeq)) != NULL while (((p = BTPDQ_FIRST(&net_bw_writeq)) != NULL
&& m_bw_bytes_out > 0)) { && m_bw_bytes_out > 0)) {
BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry); BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry);
btpd_ev_enable(&p->ioev, EV_WRITE);
p->flags &= ~PF_ON_WRITEQ; p->flags &= ~PF_ON_WRITEQ;
m_bw_bytes_out -= net_write(p, m_bw_bytes_out); m_bw_bytes_out -= net_write(p, m_bw_bytes_out);
} }
} else { } else {
while ((p = BTPDQ_FIRST(&net_bw_writeq)) != NULL) { while ((p = BTPDQ_FIRST(&net_bw_writeq)) != NULL) {
BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry); BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry);
btpd_ev_enable(&p->ioev, EV_WRITE);
p->flags &= ~PF_ON_WRITEQ; p->flags &= ~PF_ON_WRITEQ;
net_write(p, 0); net_write(p, 0);
} }
@@ -606,34 +608,49 @@ net_on_tick(void)
net_bw_tick(); net_bw_tick();
} }


void
net_read_cb(int sd, short type, void *arg)
static void
net_read_cb(struct peer *p)
{ {
struct peer *p = (struct peer *)arg;
if (net_bw_limit_in == 0) if (net_bw_limit_in == 0)
net_read(p, 0); net_read(p, 0);
else if (m_bw_bytes_in > 0) else if (m_bw_bytes_in > 0)
m_bw_bytes_in -= net_read(p, m_bw_bytes_in); m_bw_bytes_in -= net_read(p, m_bw_bytes_in);
else { else {
btpd_ev_disable(&p->ioev, EV_READ);
p->flags |= PF_ON_READQ; p->flags |= PF_ON_READQ;
BTPDQ_INSERT_TAIL(&net_bw_readq, p, rq_entry); BTPDQ_INSERT_TAIL(&net_bw_readq, p, rq_entry);
} }
} }


void
net_write_cb(int sd, short type, void *arg)
static void
net_write_cb(struct peer *p)
{ {
struct peer *p = (struct peer *)arg;
if (net_bw_limit_out == 0) if (net_bw_limit_out == 0)
net_write(p, 0); net_write(p, 0);
else if (m_bw_bytes_out > 0) else if (m_bw_bytes_out > 0)
m_bw_bytes_out -= net_write(p, m_bw_bytes_out); m_bw_bytes_out -= net_write(p, m_bw_bytes_out);
else { else {
btpd_ev_disable(&p->ioev, EV_WRITE);
p->flags |= PF_ON_WRITEQ; p->flags |= PF_ON_WRITEQ;
BTPDQ_INSERT_TAIL(&net_bw_writeq, p, wq_entry); BTPDQ_INSERT_TAIL(&net_bw_writeq, p, wq_entry);
} }
} }


void
net_io_cb(int sd, short type, void *arg)
{
switch (type) {
case EV_READ:
net_read_cb(arg);
break;
case EV_WRITE:
net_write_cb(arg);
break;
default:
abort();
}
}

void void
net_init(void) net_init(void)
{ {
@@ -641,8 +658,6 @@ net_init(void)
m_bw_bytes_in = net_bw_limit_in; m_bw_bytes_in = net_bw_limit_in;


int safe_fds = getdtablesize() * 4 / 5; int safe_fds = getdtablesize() * 4 / 5;
if (strcmp(event_get_method(), "select") == 0)
safe_fds = min(safe_fds, FD_SETSIZE * 4 / 5);
if (net_max_peers == 0 || net_max_peers > safe_fds) if (net_max_peers == 0 || net_max_peers > safe_fds)
net_max_peers = safe_fds; net_max_peers = safe_fds;


@@ -661,7 +676,5 @@ net_init(void)
listen(sd, 10); listen(sd, 10);
set_nonblocking(sd); set_nonblocking(sd);


event_set(&m_net_incoming, sd, EV_READ | EV_PERSIST,
net_connection_cb, NULL);
btpd_ev_add(&m_net_incoming, NULL);
btpd_ev_new(&m_net_incoming, sd, EV_READ, net_connection_cb, NULL);
} }

+ 1
- 2
btpd/net.h View File

@@ -31,8 +31,7 @@ int net_active(struct torrent *tp);


int net_torrent_has_peer(struct net *n, const uint8_t *id); int net_torrent_has_peer(struct net *n, const uint8_t *id);


void net_read_cb(int sd, short type, void *arg);
void net_write_cb(int sd, short type, void *arg);
void net_io_cb(int sd, short type, void *arg);


int net_connect2(struct sockaddr *sa, socklen_t salen, int *sd); int net_connect2(struct sockaddr *sa, socklen_t salen, int *sd);
int net_connect(const char *ip, int port, int *sd); int net_connect(const char *ip, int port, int *sd);


+ 1
- 2
btpd/net_types.h View File

@@ -52,8 +52,7 @@ struct peer {
size_t outq_off; size_t outq_off;
struct nb_tq outq; struct nb_tq outq;


struct event in_ev;
struct event out_ev;
struct fdev ioev;


unsigned long rate_up, rate_dwn; unsigned long rate_up, rate_dwn;
unsigned long count_up, count_dwn; unsigned long count_up, count_dwn;


+ 4
- 7
btpd/peer.c View File

@@ -23,8 +23,7 @@ peer_kill(struct peer *p)
if (p->flags & PF_ON_WRITEQ) if (p->flags & PF_ON_WRITEQ)
BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry); BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry);


btpd_ev_del(&p->in_ev);
btpd_ev_del(&p->out_ev);
btpd_ev_del(&p->ioev);
close(p->sd); close(p->sd);


nl = BTPDQ_FIRST(&p->outq); nl = BTPDQ_FIRST(&p->outq);
@@ -59,7 +58,7 @@ peer_send(struct peer *p, struct net_buf *nb)


if (BTPDQ_EMPTY(&p->outq)) { if (BTPDQ_EMPTY(&p->outq)) {
assert(p->outq_off == 0); assert(p->outq_off == 0);
btpd_ev_add(&p->out_ev, NULL);
btpd_ev_enable(&p->ioev, EV_WRITE);
p->t_wantwrite = btpd_seconds; p->t_wantwrite = btpd_seconds;
} }
BTPDQ_INSERT_TAIL(&p->outq, nl, entry); BTPDQ_INSERT_TAIL(&p->outq, nl, entry);
@@ -88,7 +87,7 @@ peer_unsend(struct peer *p, struct nb_link *nl)
BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry); BTPDQ_REMOVE(&net_bw_writeq, p, wq_entry);
p->flags &= ~PF_ON_WRITEQ; p->flags &= ~PF_ON_WRITEQ;
} else } else
btpd_ev_del(&p->out_ev);
btpd_ev_disable(&p->ioev, EV_WRITE);
} }
return 1; return 1;
} else } else
@@ -275,9 +274,7 @@ peer_create_common(int sd)


peer_set_in_state(p, SHAKE_PSTR, 28); peer_set_in_state(p, SHAKE_PSTR, 28);


event_set(&p->out_ev, p->sd, EV_WRITE, net_write_cb, p);
event_set(&p->in_ev, p->sd, EV_READ, net_read_cb, p);
btpd_ev_add(&p->in_ev, NULL);
btpd_ev_new(&p->ioev, p->sd, EV_READ, net_io_cb, p);


BTPDQ_INSERT_TAIL(&net_unattached, p, p_entry); BTPDQ_INSERT_TAIL(&net_unattached, p, p_entry);
net_npeers++; net_npeers++;


+ 2
- 3
btpd/thread_cb.c View File

@@ -15,7 +15,7 @@ struct td_cb {
BTPDQ_HEAD(td_cb_tq, td_cb); BTPDQ_HEAD(td_cb_tq, td_cb);


static int m_td_rd, m_td_wr; static int m_td_rd, m_td_wr;
static struct event m_td_ev;
static struct fdev m_td_ev;
static struct td_cb_tq m_td_cbs = BTPDQ_HEAD_INITIALIZER(m_td_cbs); static struct td_cb_tq m_td_cbs = BTPDQ_HEAD_INITIALIZER(m_td_cbs);
static pthread_mutex_t m_td_lock; static pthread_mutex_t m_td_lock;


@@ -82,6 +82,5 @@ td_init(void)
if ((err = pthread_mutex_init(&m_td_lock, NULL)) != 0) if ((err = pthread_mutex_init(&m_td_lock, NULL)) != 0)
btpd_err("Couldn't create mutex (%s).\n", strerror(err)); btpd_err("Couldn't create mutex (%s).\n", strerror(err));


event_set(&m_td_ev, m_td_rd, EV_READ|EV_PERSIST, td_cb, NULL);
btpd_ev_add(&m_td_ev, NULL);
btpd_ev_new(&m_td_ev, m_td_rd, EV_READ, td_cb, NULL);
} }

+ 12
- 12
btpd/tracker_req.c View File

@@ -2,8 +2,8 @@


#define REQ_DELAY 1 #define REQ_DELAY 1
#define STOP_ERRORS 5 #define STOP_ERRORS 5
#define REQ_TIMEOUT (& (struct timeval) { 120, 0 })
#define RETRY_WAIT (& (struct timeval) { rand_between(35, 70), 0 })
#define REQ_TIMEOUT (& (struct timespec) { 120, 0 })
#define RETRY_WAIT (& (struct timespec) { rand_between(35, 70), 0 })


long tr_key; long tr_key;


@@ -24,7 +24,7 @@ struct tracker {
int tier, url; int tier, url;
struct mi_announce *ann; struct mi_announce *ann;
void *req; void *req;
struct event timer;
struct timeout timer;
}; };


typedef struct _dummy *(*request_fun_t)(struct torrent *, enum tr_event, typedef struct _dummy *(*request_fun_t)(struct torrent *, enum tr_event,
@@ -97,7 +97,7 @@ static void
tr_set_stopped(struct torrent *tp) tr_set_stopped(struct torrent *tp)
{ {
struct tracker *tr = tp->tr; struct tracker *tr = tp->tr;
btpd_ev_del(&tr->timer);
btpd_timer_del(&tr->timer);
tr->ttype = TIMER_NONE; tr->ttype = TIMER_NONE;
if (tr->req != NULL) if (tr->req != NULL)
tr_cancel(tr); tr_cancel(tr);
@@ -116,8 +116,8 @@ tr_send(struct torrent *tp, enum tr_event event)
if (m_tlast_req > btpd_seconds - REQ_DELAY) { if (m_tlast_req > btpd_seconds - REQ_DELAY) {
m_tnext_req = max(m_tnext_req, m_tlast_req) + REQ_DELAY; m_tnext_req = max(m_tnext_req, m_tlast_req) + REQ_DELAY;
tr->ttype = TIMER_RETRY; tr->ttype = TIMER_RETRY;
btpd_ev_add(&tr->timer,
(& (struct timeval) { m_tnext_req - btpd_seconds, 0 }));
btpd_timer_add(&tr->timer,
(& (struct timespec) { m_tnext_req - btpd_seconds, 0 }));
return; return;
} }


@@ -130,11 +130,11 @@ tr_send(struct torrent *tp, enum tr_event event)
} }
next_url(tr); next_url(tr);
tr->ttype = TIMER_RETRY; tr->ttype = TIMER_RETRY;
btpd_ev_add(&tr->timer, (& (struct timeval) { 5, 0 }));
btpd_timer_add(&tr->timer, (& (struct timespec) { 5, 0 }));
} else { } else {
m_tlast_req = btpd_seconds; m_tlast_req = btpd_seconds;
tr->ttype = TIMER_TIMEOUT; tr->ttype = TIMER_TIMEOUT;
btpd_ev_add(&tr->timer, REQ_TIMEOUT);
btpd_timer_add(&tr->timer, REQ_TIMEOUT);
} }
} }


@@ -151,11 +151,11 @@ tr_result(struct torrent *tp, enum tr_res res, int interval)
tr->interval = interval; tr->interval = interval;
tr->nerrors = 0; tr->nerrors = 0;
tr->ttype = TIMER_INTERVAL; tr->ttype = TIMER_INTERVAL;
btpd_ev_add(&tr->timer, (& (struct timeval) { tr->interval, 0}));
btpd_timer_add(&tr->timer, (& (struct timespec) { tr->interval, 0}));
} else { } else {
tr->nerrors++; tr->nerrors++;
tr->ttype = TIMER_RETRY; tr->ttype = TIMER_RETRY;
btpd_ev_add(&tr->timer, RETRY_WAIT);
btpd_timer_add(&tr->timer, RETRY_WAIT);
next_url(tr); next_url(tr);
} }
} }
@@ -193,7 +193,7 @@ tr_create(struct torrent *tp, const char *mi)
tp->tr = btpd_calloc(1, sizeof(*tp->tr)); tp->tr = btpd_calloc(1, sizeof(*tp->tr));
if ((tp->tr->ann = mi_announce(mi)) == NULL) if ((tp->tr->ann = mi_announce(mi)) == NULL)
btpd_err("Out of memory.\n"); btpd_err("Out of memory.\n");
evtimer_set(&tp->tr->timer, timer_cb, tp);
timer_init(&tp->tr->timer, timer_cb, tp);
return 0; return 0;
} }


@@ -202,7 +202,7 @@ tr_kill(struct torrent *tp)
{ {
struct tracker *tr = tp->tr; struct tracker *tr = tp->tr;
tp->tr = NULL; tp->tr = NULL;
btpd_ev_del(&tr->timer);
btpd_timer_del(&tr->timer);
if (tr->req != NULL) if (tr->req != NULL)
tr_cancel(tr); tr_cancel(tr);
mi_free_announce(tr->ann); mi_free_announce(tr->ann);


+ 5
- 5
btpd/upload.c View File

@@ -1,8 +1,8 @@
#include "btpd.h" #include "btpd.h"


#define CHOKE_INTERVAL (& (struct timeval) { 10, 0 })
#define CHOKE_INTERVAL (& (struct timespec) { 10, 0 })


static struct event m_choke_timer;
static struct timeout m_choke_timer;
static unsigned m_npeers; static unsigned m_npeers;
static struct peer_tq m_peerq = BTPDQ_HEAD_INITIALIZER(m_peerq); static struct peer_tq m_peerq = BTPDQ_HEAD_INITIALIZER(m_peerq);
static int m_max_uploads; static int m_max_uploads;
@@ -110,7 +110,7 @@ shuffle_optimists(void)
static void static void
choke_cb(int sd, short type, void *arg) choke_cb(int sd, short type, void *arg)
{ {
btpd_ev_add(&m_choke_timer, CHOKE_INTERVAL);
btpd_timer_add(&m_choke_timer, CHOKE_INTERVAL);
static int cb_count = 0; static int cb_count = 0;
cb_count++; cb_count++;
if (cb_count % 3 == 0) if (cb_count % 3 == 0)
@@ -190,6 +190,6 @@ ul_init(void)
m_max_uploads = 5 + (net_bw_limit_out / (100 << 10)); m_max_uploads = 5 + (net_bw_limit_out / (100 << 10));
} }


evtimer_set(&m_choke_timer, choke_cb, NULL);
btpd_ev_add(&m_choke_timer, CHOKE_INTERVAL);
timer_init(&m_choke_timer, choke_cb, NULL);
btpd_timer_add(&m_choke_timer, CHOKE_INTERVAL);
} }

+ 31
- 4
btpd/util.c View File

@@ -22,19 +22,46 @@ btpd_calloc(size_t nmemb, size_t size)
} }


void void
btpd_ev_add(struct event *ev, struct timeval *tv)
btpd_ev_new(struct fdev *ev, int fd, uint16_t flags, evloop_cb_t cb, void *arg)
{ {
if (event_add(ev, tv) != 0)
if (fdev_new(ev, fd, flags, cb, arg) != 0)
btpd_err("Failed to add event (%s).\n", strerror(errno)); btpd_err("Failed to add event (%s).\n", strerror(errno));
} }


void void
btpd_ev_del(struct event *ev)
btpd_ev_del(struct fdev *ev)
{ {
if (event_del(ev) != 0)
if (fdev_del(ev) != 0)
btpd_err("Failed to remove event (%s).\n", strerror(errno)); btpd_err("Failed to remove event (%s).\n", strerror(errno));
} }


void
btpd_ev_enable(struct fdev *ev, uint16_t flags)
{
if (fdev_enable(ev, flags) != 0)
btpd_err("Failed to enable event (%s).\n", strerror(errno));
}

void
btpd_ev_disable(struct fdev *ev, uint16_t flags)
{
if (fdev_disable(ev, flags) != 0)
btpd_err("Failed to disable event (%s).\n", strerror(errno));
}

void
btpd_timer_add(struct timeout *to, struct timespec *ts)
{
if (timer_add(to, ts) != 0)
btpd_err("Failed to add timeout (%s).\n", strerror(errno));
}

void
btpd_timer_del(struct timeout *to)
{
timer_del(to);
}

static const char * static const char *
logtype_str(uint32_t type) logtype_str(uint32_t type)
{ {


Loading…
Cancel
Save