Browse Source

Improve the torrent stop and btpd shutdown sequences.

Torrents are now considered stopped and may be restarted even if
the stop event haven't been sent the trackers yet. The same holds
for the del and add commands.

A btpd process in shutdown mode that only have stopped torrents,
but is still sending the stop event to trackers, will release
resources that would block a new btpd to start. It will the
silently exit when it's finished with the trackers.

This also makes the timeout parameter for shutdown unnecessary.
master
Richard Nyberg 16 years ago
parent
commit
438881f16f
12 changed files with 179 additions and 88 deletions
  1. +28
    -22
      btpd/btpd.c
  2. +1
    -1
      btpd/btpd.h
  3. +33
    -20
      btpd/cli_if.c
  4. +3
    -3
      btpd/main.c
  5. +11
    -0
      btpd/net.c
  6. +16
    -2
      btpd/tlib.c
  7. +3
    -0
      btpd/tlib.h
  8. +71
    -19
      btpd/torrent.c
  9. +7
    -1
      btpd/torrent.h
  10. +3
    -14
      cli/kill.c
  11. +2
    -5
      misc/btpd_if.c
  12. +1
    -1
      misc/btpd_if.h

+ 28
- 22
btpd/btpd.c View File

@@ -5,9 +5,9 @@

static uint8_t m_peer_id[20];
static struct timeout m_heartbeat;
static struct timeout m_grace_timer;
static int m_signal;
static int m_shutdown;
static int m_ghost;

long btpd_seconds;

@@ -18,29 +18,35 @@ btpd_exit(int code)
exit(code);
}

extern int pidfd;
void ipc_shutdown(void);
void net_shutdown(void);

static void
grace_cb(int fd, short type, void *arg)
death_procedure(void)
{
struct torrent *tp;
BTPDQ_FOREACH(tp, torrent_get_all(), entry)
torrent_stop(tp, 0);
assert(m_shutdown);
if (torrent_count() == 0)
btpd_exit(0);
if (!m_ghost && torrent_count() == torrent_ghosts()) {
btpd_log(BTPD_L_BTPD, "Entering pre exit mode. Bye!\n");
fclose(stderr);
fclose(stdout);
net_shutdown();
ipc_shutdown();
close(pidfd);
m_ghost = 1;
}
}

void
btpd_shutdown(int grace_seconds)
btpd_shutdown(void)
{
if (torrent_count() == 0)
btpd_exit(0);
else {
struct torrent *tp;
m_shutdown = 1;
BTPDQ_FOREACH(tp, torrent_get_all(), entry)
if (tp->state != T_STOPPING)
torrent_stop(tp, 0);
if (grace_seconds >= 0)
btpd_timer_add(&m_grace_timer,
(& (struct timespec){ grace_seconds, 0 }));
}
m_shutdown = 1;
struct torrent *tp, *next;
BTPDQ_FOREACH_MUTABLE(tp, torrent_get_all(), entry, next)
torrent_stop(tp, 0);
death_procedure();
}

int btpd_is_stopping(void)
@@ -69,11 +75,12 @@ heartbeat_cb(int fd, short type, void *arg)
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)
btpd_shutdown();
}
if (m_shutdown && torrent_count() == 0)
btpd_exit(0);
if (m_shutdown)
death_procedure();
}

void tr_init(void);
@@ -123,7 +130,6 @@ btpd_init(void)
tr_init();
tlib_init();

timer_init(&m_grace_timer, grace_cb, NULL);
timer_init(&m_heartbeat, heartbeat_cb, NULL);
btpd_timer_add(&m_heartbeat, (& (struct timespec) { 1, 0 }));
}

+ 1
- 1
btpd/btpd.h View File

@@ -82,7 +82,7 @@ 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(void);
int btpd_is_stopping(void);

const uint8_t *btpd_get_peer_id(void);


+ 33
- 20
btpd/cli_if.c View File

@@ -8,6 +8,7 @@ struct cli {
struct fdev read;
};

static int m_listen_sd;
static struct fdev m_cli_incoming;

static int
@@ -135,6 +136,8 @@ write_ans(struct iobuf *iob, struct tlib *tl, enum ipc_tval val)
case T_LEECH:
ts = IPC_TSTATE_LEECH;
break;
case T_GHOST:
break;
}
}
iobuf_print(iob, "i%de", ts);
@@ -185,9 +188,10 @@ cmd_tget(struct cli *cli, int argc, const char *args)
struct tlib *tlv[tlib_count()];
tlib_put_all(tlv);
for (int i = 0; i < sizeof(tlv) / sizeof(tlv[0]); i++) {
if ((from == IPC_TWC_ALL ||
(tlv[i]->tp == NULL && from == IPC_TWC_INACTIVE) ||
(tlv[i]->tp != NULL && from == IPC_TWC_ACTIVE))) {
if (!torrent_haunting(tlv[i]) && (
from == IPC_TWC_ALL ||
(!torrent_active(tlv[i]) && from == IPC_TWC_INACTIVE) ||
(torrent_active(tlv[i]) && from == IPC_TWC_ACTIVE))) {
iobuf_swrite(&iob, "l");
for (int k = 0; k < nkeys; k++)
write_ans(&iob, tlv[i], opts[k]);
@@ -206,7 +210,7 @@ cmd_tget(struct cli *cli, int argc, const char *args)
free(opts);
return IPC_COMMERR;
}
if (tl != NULL) {
if (tl != NULL && !torrent_haunting(tl)) {
iobuf_swrite(&iob, "l");
for (int i = 0; i < nkeys; i++)
write_ans(&iob, tl, opts[i]);
@@ -248,10 +252,15 @@ cmd_add(struct cli *cli, int argc, const char *args)
content[csize] = '\0';

tl = tlib_by_hash(mi_info_hash(mi, hash));
if (tl != NULL)
if (tl != NULL && !torrent_haunting(tl))
return write_code_buffer(cli, IPC_ETENTEXIST);
tl = tlib_add(hash, mi, mi_size, content,
benc_dget_str(args, "name", NULL));
if (tl != NULL) {
tl = tlib_readd(tl, hash, mi, mi_size, content,
benc_dget_str(args, "name", NULL));
} else {
tl = tlib_add(hash, mi, mi_size, content,
benc_dget_str(args, "name", NULL));
}
return write_add_buffer(cli, tl->num);
}

@@ -270,7 +279,7 @@ cmd_del(struct cli *cli, int argc, const char *args)
else
return IPC_COMMERR;

if (tl == NULL)
if (tl == NULL || torrent_haunting(tl))
ret = write_code_buffer(cli, IPC_ENOTENT);
else {
ret = write_code_buffer(cli, IPC_OK);
@@ -300,9 +309,9 @@ cmd_start(struct cli *cli, int argc, const char *args)
else
return IPC_COMMERR;

if (tl == NULL)
if (tl == NULL || torrent_haunting(tl))
code = IPC_ENOTENT;
else if (tl->tp != NULL)
else if (!torrent_startable(tl))
code = IPC_ETACTIVE;
else
if ((code = torrent_start(tl)) == IPC_OK)
@@ -324,9 +333,9 @@ cmd_stop(struct cli *cli, int argc, const char *args)
else
return IPC_COMMERR;

if (tl == NULL)
if (tl == NULL || torrent_haunting(tl))
return write_code_buffer(cli, IPC_ENOTENT);
else if (tl->tp == NULL)
else if (!torrent_active(tl))
return write_code_buffer(cli, IPC_ETINACTIVE);
else {
// Stopping a torrent may trigger exit so we need to reply before.
@@ -340,12 +349,11 @@ cmd_stop(struct cli *cli, int argc, const char *args)
static int
cmd_stop_all(struct cli *cli, int argc, const char *args)
{
struct torrent *tp;
struct torrent *tp, *next;
int ret = write_code_buffer(cli, IPC_OK);
active_clear();
BTPDQ_FOREACH(tp, torrent_get_all(), entry)
if (tp->state != T_STOPPING)
torrent_stop(tp, 0);
BTPDQ_FOREACH_MUTABLE(tp, torrent_get_all(), entry, next)
torrent_stop(tp, 0);
return ret;
}

@@ -354,11 +362,8 @@ cmd_die(struct cli *cli, int argc, const char *args)
{
int err = write_code_buffer(cli, IPC_OK);
if (!btpd_is_stopping()) {
int grace_seconds = -1;
if (argc == 1 && benc_isint(args))
grace_seconds = benc_int(args, NULL);
btpd_log(BTPD_L_BTPD, "Someone wants me dead.\n");
btpd_shutdown(grace_seconds);
btpd_shutdown();
}
return err;
}
@@ -449,6 +454,13 @@ client_connection_cb(int sd, short type, void *arg)
btpd_ev_new(&cli->read, cli->sd, EV_READ, cli_read_cb, cli);
}

void
ipc_shutdown(void)
{
btpd_ev_del(&m_cli_incoming);
close(m_listen_sd);
}

void
ipc_init(void)
{
@@ -477,4 +489,5 @@ ipc_init(void)
set_nonblocking(sd);

btpd_ev_new(&m_cli_incoming, sd, EV_READ, client_connection_cb, NULL);
m_listen_sd = sd;
}

+ 3
- 3
btpd/main.c View File

@@ -5,6 +5,7 @@

int btpd_daemon_phase = 2;
int first_btpd_comm[2];
int pidfd;

void
first_btpd_exit(char code)
@@ -15,7 +16,7 @@ first_btpd_exit(char code)
}

static void
writepid(int pidfd)
writepid(void)
{
int nw;
char pidtxt[100];
@@ -28,7 +29,6 @@ static void
setup_daemon(int daemonize, const char *dir)
{
char c;
int pidfd;
pid_t pid;
struct timespec ts;

@@ -91,7 +91,7 @@ setup_daemon(int daemonize, const char *dir)
if (lockf(pidfd, F_TLOCK, 0) == -1)
btpd_err("Another instance of btpd is probably running in %s.\n", dir);

writepid(pidfd);
writepid();
}

static void


+ 11
- 0
btpd/net.c View File

@@ -14,6 +14,7 @@ struct net_listener {
struct fdev ev;
};

static int m_nlisteners;
static struct net_listener *m_net_listeners;

unsigned net_npeers;
@@ -668,6 +669,15 @@ net_af_spec(void)
return AF_INET6;
}

void
net_shutdown(void)
{
for (int i = 0; i < m_nlisteners; i++) {
btpd_ev_del(&m_net_listeners[i].ev);
close(m_net_listeners[i].sd);
}
}

void
net_init(void)
{
@@ -699,6 +709,7 @@ net_init(void)
net_ipv6 = found_ipv6;
if (!net_ipv4 && !net_ipv6)
btpd_err("no usable address found. wrong use of -4/-6 perhaps.\n");
m_nlisteners = count;
m_net_listeners = btpd_calloc(count, sizeof(*m_net_listeners));
for (ai = res; ai != NULL; ai = ai->ai_next) {
count--;


+ 16
- 2
btpd/tlib.c View File

@@ -65,7 +65,7 @@ tlib_del(struct tlib *tl)
char path[PATH_MAX];
DIR *dir;
struct dirent *de;
assert(tl->tp == NULL);
assert(tl->tp == NULL || tl->tp->state == T_GHOST);
snprintf(path, PATH_MAX, "torrents/%s", bin2hex(tl->hash, relpath, 20));
if ((dir = opendir(path)) != NULL) {
while ((de = readdir(dir)) != NULL) {
@@ -78,7 +78,8 @@ tlib_del(struct tlib *tl)
}
snprintf(path, PATH_MAX, "torrents/%s", relpath);
remove(path);
tlib_kill(tl);
if (tl->tp == NULL)
tlib_kill(tl);
return 0;
}

@@ -257,6 +258,19 @@ tlib_add(const uint8_t *hash, const char *mi, size_t mi_size,
return tl;
}

struct tlib *
tlib_readd(struct tlib *tl, const uint8_t *hash, const char *mi,
size_t mi_size, const char *content, char *name)
{
struct tlib *tln;
struct torrent *tp = tl->tp;
tp->delete = 0;
tlib_kill(tl);
tln = tlib_add(hash, mi, mi_size, content, name);
tln->tp = tp;
return tln;
}

static int
num_test(const void *k1, const void *k2)
{


+ 3
- 0
btpd/tlib.h View File

@@ -26,7 +26,10 @@ void tlib_put_all(struct tlib **v);

struct tlib *tlib_add(const uint8_t *hash, const char *mi, size_t mi_size,
const char *content, char *name);
struct tlib *tlib_readd(struct tlib *tl, const uint8_t *hash, const char *mi,
size_t mi_size, const char *content, char *name);
int tlib_del(struct tlib *tl);
void tlib_kill(struct tlib *tl);

void tlib_update_info(struct tlib *tl, int only_file);



+ 71
- 19
btpd/torrent.c View File

@@ -4,6 +4,7 @@

#define SAVE_INTERVAL 300

static unsigned m_nghosts;
static unsigned m_ntorrents;
static struct torrent_tq m_torrents = BTPDQ_HEAD_INITIALIZER(m_torrents);

@@ -22,6 +23,12 @@ torrent_count(void)
return m_ntorrents;
}

unsigned
torrent_ghosts(void)
{
return m_nghosts;
}

struct torrent *
torrent_by_num(unsigned num)
{
@@ -71,12 +78,41 @@ torrent_block_size(struct torrent *tp, uint32_t piece, uint32_t nblocks,
}
}

static void
torrent_kill(struct torrent *tp)
{
assert(m_ntorrents > 0);
assert(!(net_active(tp) || cm_active(tp)));
if (tp->state == T_GHOST)
m_nghosts--;
m_ntorrents--;
BTPDQ_REMOVE(&m_torrents, tp, entry);
if (tp->delete)
tlib_kill(tp->tl);
else
tp->tl->tp = NULL;
tr_kill(tp);
net_kill(tp);
cm_kill(tp);
mi_free_files(tp->nfiles, tp->files);
if (m_savetp == tp)
if ((m_savetp = BTPDQ_NEXT(tp, entry)) == NULL)
m_savetp = BTPDQ_FIRST(&m_torrents);
free(tp);
}

enum ipc_err
torrent_start(struct tlib *tl)
{
struct torrent *tp;
char *mi;

if (tl->tp != NULL) {
assert(torrent_startable(tl));
torrent_kill(tl->tp);
tl->tp = NULL;
}

if (tl->dir == NULL)
return IPC_EBADTENT;

@@ -111,25 +147,16 @@ torrent_start(struct tlib *tl)
return IPC_OK;
}

static void
torrent_kill(struct torrent *tp)
static
void become_ghost(struct torrent *tp)
{
btpd_log(BTPD_L_BTPD, "Stopped torrent '%s'.\n", torrent_name(tp));
assert(m_ntorrents > 0);
assert(!(tr_active(tp) || net_active(tp) || cm_active(tp)));
m_ntorrents--;
BTPDQ_REMOVE(&m_torrents, tp, entry);
tp->tl->tp = NULL;
tp->state = T_GHOST;
if (tp->delete)
tlib_del(tp->tl);
tr_kill(tp);
net_kill(tp);
cm_kill(tp);
mi_free_files(tp->nfiles, tp->files);
if (m_savetp == tp)
if ((m_savetp = BTPDQ_NEXT(tp, entry)) == NULL)
m_savetp = BTPDQ_FIRST(&m_torrents);
free(tp);
else
tlib_update_info(tp->tl, 0);
m_nghosts++;
}

void
@@ -148,10 +175,13 @@ torrent_stop(struct torrent *tp, int delete)
tr_stop(tp);
if (cm_active(tp))
cm_stop(tp);
if (!delete)
tlib_update_info(tp->tl, 0);
if (!cm_active(tp)) {
become_ghost(tp);
if (!tr_active(tp))
torrent_kill(tp);
}
break;
case T_STOPPING:
default:
break;
}
}
@@ -187,7 +217,11 @@ torrent_on_tick(struct torrent *tp)
}
break;
case T_STOPPING:
if (!(cm_active(tp) || tr_active(tp)))
if (cm_active(tp))
break;
become_ghost(tp);
case T_GHOST:
if (!tr_active(tp))
torrent_kill(tp);
break;
default:
@@ -213,3 +247,21 @@ torrent_on_tick_all(void)
}
}
}

int
torrent_active(struct tlib *tl)
{
return tl->tp != NULL && tl->tp->state != T_GHOST;
}

int
torrent_startable(struct tlib *tl)
{
return tl->tp == NULL || (tl->tp->state == T_GHOST && !tl->tp->delete);
}

int
torrent_haunting(struct tlib *tl)
{
return tl->tp != NULL && tl->tp->delete && tl->tp->state == T_GHOST;
}

+ 7
- 1
btpd/torrent.h View File

@@ -8,7 +8,8 @@ enum torrent_state {
T_STARTING,
T_LEECH,
T_SEED,
T_STOPPING
T_STOPPING,
T_GHOST
};

struct torrent {
@@ -34,6 +35,7 @@ struct torrent {
BTPDQ_HEAD(torrent_tq, torrent);

unsigned torrent_count(void);
unsigned torrent_ghosts(void);
const struct torrent_tq *torrent_get_all(void);
struct torrent *torrent_by_num(unsigned num);
struct torrent *torrent_by_hash(const uint8_t *hash);
@@ -41,6 +43,10 @@ struct torrent *torrent_by_hash(const uint8_t *hash);
enum ipc_err torrent_start(struct tlib *tl);
void torrent_stop(struct torrent *tp, int delete);

int torrent_active(struct tlib *tl);
int torrent_haunting(struct tlib *tl);
int torrent_startable(struct tlib *tl);

off_t torrent_piece_size(struct torrent *tp, uint32_t piece);
uint32_t torrent_piece_blocks(struct torrent *tp, uint32_t piece);
uint32_t torrent_block_size(struct torrent *tp, uint32_t piece,


+ 3
- 14
cli/kill.c View File

@@ -6,12 +6,7 @@ usage_kill(void)
printf(
"Shutdown btpd.\n"
"\n"
"Usage: kill [seconds]\n"
"\n"
"Arguments:\n"
"seconds\n"
"\tThe number of seconds btpd waits before giving up on unresponsive\n"
"\ttrackers.\n"
"Usage: kill\n"
"\n"
);
exit(1);
@@ -20,18 +15,12 @@ usage_kill(void)
void
cmd_kill(int argc, char **argv)
{
int seconds = -1;
enum ipc_err code;
char *endptr;

if (argc == 2) {
seconds = strtol(argv[1], &endptr, 10);
if (strlen(argv[1]) > endptr - argv[1] || seconds < 0)
usage_kill();
} else if (argc > 2)
if (argc > 1)
usage_kill();

btpd_connect();
if ((code = btpd_die(ipc, seconds)) != 0)
if ((code = btpd_die(ipc)) != 0)
diemsg("command failed (%s).\n", ipc_strerror(code));
}

+ 2
- 5
misc/btpd_if.c View File

@@ -155,13 +155,10 @@ ipc_buf_req_code(struct ipc *ipc, struct iobuf *iob)
}

enum ipc_err
btpd_die(struct ipc *ipc, int seconds)
btpd_die(struct ipc *ipc)
{
struct iobuf iob = iobuf_init(16);
if (seconds >= 0)
iobuf_print(&iob, "l3:diei%dee", seconds);
else
iobuf_swrite(&iob, "l3:diee");
iobuf_swrite(&iob, "l3:diee");
return ipc_buf_req_code(ipc, &iob);
}



+ 1
- 1
misc/btpd_if.h View File

@@ -80,7 +80,7 @@ enum ipc_err btpd_del(struct ipc *ipc, struct ipc_torrent *tp);
enum ipc_err btpd_start(struct ipc *ipc, struct ipc_torrent *tp);
enum ipc_err btpd_stop(struct ipc *ipc, struct ipc_torrent *tp);
enum ipc_err btpd_stop_all(struct ipc *ipc);
enum ipc_err btpd_die(struct ipc *ipc, int seconds);
enum ipc_err btpd_die(struct ipc *ipc);
enum ipc_err btpd_get(struct ipc *ipc, enum ipc_dval *keys, size_t nkeys,
tget_cb_t cb, void *arg);
enum ipc_err btpd_tget(struct ipc *ipc, struct ipc_torrent *tps, size_t ntps,


Loading…
Cancel
Save