[Messy Patch] XOVER Command Pipelining
Jan Andres <[email protected]> Sun, 29 Jul 2012 22:59:19 +0200
| Newsgroups | gmane.network.slrn.user |
|---|---|
| Message-ID | <[email protected]> |
Hello,
I've noticed many times that slrn can be a bit slow when opening busy
groups where it needs to send many XOVER commands to the server; in my
case the server isn't actually slow to reply. Rather, the issue is
network latency and slrn's consuming the response to each XOVER
command before sending out the next one.
RFC 3977 explicitly allows command pipelining in NNTP so I was wondering
if that could help speed things up. And indeed it does. I hacked up a
little proof-of-concept patch and found that slrn can now open all my
groups within less than a second, which is at least a 10-fold
improvement.
The basic idea is initially to prepare a buffer containing all the
desired XOVER commands and extend the select() loop in wait_for_input()
so it will asynchronously send out the buffer's contents along with the
processing of the replies.
The patch is just a quick hack, it has basically no error handling and
it breaks a lot of things (including SSL), but it should help illustrate
the basic idea. I hope I'll have some time to get it into a more usable
state.
Cheers,
Jan
diff --git a/src/art.c b/src/art.c
index 513858a..f2dd345 100644
--- a/src/art.c
+++ b/src/art.c
@@ -5687,7 +5687,7 @@ static int get_add_headers (NNTP_Artnum_Type min, NNTP_Artnum_Type max) /*{{{*/
/*}}}*/
/* gets the headers of article number min-max, decrementing *totalp for each
* downloaded article */
-static int get_headers (NNTP_Artnum_Type min, NNTP_Artnum_Type max, NNTP_Artnum_Type *totalp) /*{{{*/
+static int get_headers (const Slrn_Artnum_List_Type *artnums, NNTP_Artnum_Type *totalp) /*{{{*/
{
Slrn_Header_Type *h;
/* int percent, last_percent, dpercent, */
@@ -5696,7 +5696,9 @@ static int get_headers (NNTP_Artnum_Type min, NNTP_Artnum_Type max, NNTP_Artnum_
int reads_per_update;
int num_processed;
int num, err;
+ //int expect_replies = slrn_artnum_list_len (artnums);
Slrn_XOver_Type xov;
+ Slrn_Artnum_Entry_Type *entry = artnums->first;
if (total == 0)
return 0;
@@ -5709,7 +5711,13 @@ static int get_headers (NNTP_Artnum_Type min, NNTP_Artnum_Type max, NNTP_Artnum_
/* slrn_set_suspension (1); */
- err = slrn_open_xover (min, max);
+ slrn_send_xover (artnums);
+
+ num_processed = 0;
+
+ while (entry != NULL) { /* XXX */
+ err = slrn_open_xover_reply (entry->min, entry->max);
+ fprintf (stderr, "got status %d\n", err);
if (err != OK_XOVER)
{
if ((err == ERR_NOCRNT) || /* no articles in the range */
@@ -5719,8 +5727,8 @@ static int get_headers (NNTP_Artnum_Type min, NNTP_Artnum_Type max, NNTP_Artnum_
return -1;
}
- num_processed = 0;
- expected_num = min;
+ expected_num = entry->min;
+
num = Total_Num_Headers + Number_Killed;
while (slrn_read_xover(&xov) > 0)
{
@@ -5775,20 +5783,23 @@ static int get_headers (NNTP_Artnum_Type min, NNTP_Artnum_Type max, NNTP_Artnum_
num_processed++;
}
- slrn_close_xover ();
-
- if (expected_num != max + 1)
+ if (expected_num != entry->max + 1)
{
NNTP_Artnum_Type bad_num;
- total -= (max - expected_num) + 1;
+ total -= (entry->max - expected_num) + 1;
- for (bad_num = expected_num; bad_num <= max; bad_num++)
+ for (bad_num = expected_num; bad_num <= entry->max; bad_num++)
add_to_missing_article_list (bad_num);
}
- if (-1 == get_add_headers (min, max))
- return -1;
+ entry = entry->next;
+ } /* XXX */
+
+ slrn_close_xover ();
+
+ /*if (-1 == get_add_headers (min, max))
+ return -1;*/
/* slrn_set_suspension (0); */
*totalp = total;
@@ -6023,10 +6034,15 @@ static int find_children_headers (Slrn_Header_Type *parent, /*{{{*/
{
Slrn_XOver_Type xov;
Slrn_Header_Type *h = NULL;
+ Slrn_Artnum_List_Type list;
+ int rc;
id = id_array[i];
- if (OK_XOVER != slrn_open_xover (id, id))
+ slrn_artnum_list_init (&list); /* XXX clear this when done */
+ slrn_artnum_list_append (&list, id, id);
+ slrn_send_xover (&list);
+ if (OK_XOVER != slrn_open_xover_reply (id, id))
break;
/* This will loop once. */
@@ -7837,14 +7853,21 @@ int slrn_select_article_mode (Slrn_Group_Type *g, NNTP_Artnum_Type all, int scor
if (all > 0)
{
+ Slrn_Artnum_List_Type list;
+ slrn_artnum_list_init (&list);
min = Slrn_Server_Max - all + 1;
if (min < Slrn_Server_Min) min = Slrn_Server_Min;
- status = get_headers (min, Slrn_Server_Max, &all);
+ slrn_artnum_list_append (&list, min, Slrn_Server_Max);
+ status = get_headers (&list, &all);
+ slrn_artnum_list_clear (&list);
if (status != -1)
mark_ranges (r, HEADER_READ);
}
else
{
+ Slrn_Artnum_List_Type list;
+ slrn_artnum_list_init (&list);
+
if ((all < 0) && (r != NULL))
{
NNTP_Artnum_Type unread;
@@ -7928,16 +7951,10 @@ int slrn_select_article_mode (Slrn_Group_Type *g, NNTP_Artnum_Type all, int scor
min = smin;
max = r->min - 1;
- status = get_headers (min, max, &all);
-
- if (status == -1)
- break;
+ slrn_artnum_list_append (&list, min, max);
- if (status == 0)
- {
- Slrn_Groups_Dirty = 1;
- r->min = min;
- }
+ Slrn_Groups_Dirty = 1;
+ r->min = min;
smin = r->max + 1;
}
@@ -7950,8 +7967,11 @@ int slrn_select_article_mode (Slrn_Group_Type *g, NNTP_Artnum_Type all, int scor
if (smin <= smax)
{
- status = get_headers (smin, smax, &all);
+ slrn_artnum_list_append (&list, smin, smax);
}
+
+ status = get_headers (&list, &all);
+ slrn_artnum_list_clear (&list);
}
slrn_close_add_xover (1);
diff --git a/src/nntp.c b/src/nntp.c
index 1e2c2bc..356f726 100644
--- a/src/nntp.c
+++ b/src/nntp.c
@@ -340,11 +340,23 @@ static int _nntp_po_printf (char *fmt, ...)
return retval;
}
-static int _nntp_xover_cmd (NNTP_Artnum_Type min, NNTP_Artnum_Type max)
+static int _nntp_send_xover_cmd (const Slrn_Artnum_List_Type *artnums)
{
int status;
- if (OK_XOVER == (status = nntp_xover_cmd (NNTP_Server, min, max)))
+ if (0 == (status = nntp_send_xover_cmd (NNTP_Server, artnums)))
+ {
+ _NNTP_Abort_On_Disconnection = -1;
+ }
+
+ return status;
+}
+
+static int _nntp_xover_reply_cmd (int min, int max)
+{
+ int status;
+
+ if (OK_XOVER == (status = nntp_get_server_response (NNTP_Server)))
{
_NNTP_Abort_On_Disconnection = -1;
}
@@ -442,7 +454,8 @@ static int nntp_init_objects (void)
NNTP_Server_Obj.sv_has_xover = 0;
NNTP_Server_Obj.sv_has_xhdr = -1;
- NNTP_Server_Obj.sv_nntp_xover = _nntp_xover_cmd;
+ NNTP_Server_Obj.sv_nntp_send_xover = _nntp_send_xover_cmd;
+ NNTP_Server_Obj.sv_nntp_xover_reply = _nntp_xover_reply_cmd;
NNTP_Server_Obj.sv_nntp_xhdr = _nntp_xhdr_cmd;
NNTP_Server_Obj.sv_nntp_head = _nntp_head_cmd;
NNTP_Server_Obj.sv_nntp_next = _nntp_next_cmd;
diff --git a/src/nntplib.c b/src/nntplib.c
index 2f59f36..9c00531 100644
--- a/src/nntplib.c
+++ b/src/nntplib.c
@@ -397,6 +397,11 @@ int nntp_start_server_vcmd (NNTP_Type *s, char *fmt, ...)
return nntp_start_server_cmd (s, buf);
}
+int nntp_server_pipelined_cmds (NNTP_Type *s, const char *input, unsigned input_len)
+{
+ sltcp_queue_for_write (s->tcp, input, input_len);
+}
+
int nntp_close_server (NNTP_Type *s)
{
if (s == NULL)
@@ -865,9 +870,26 @@ int nntp_head_cmd (NNTP_Type *s, NNTP_Artnum_Type n, char *msgid, NNTP_Artnum_Ty
return status;
}
-int nntp_xover_cmd (NNTP_Type *s, NNTP_Artnum_Type min, NNTP_Artnum_Type max)
+int nntp_send_xover_cmd (NNTP_Type *s, const Slrn_Artnum_List_Type *artnums)
{
- return nntp_server_vcmd (s, "XOVER " NNTP_FMT_ARTRANGE, min, max);
+ char buf[1048576];
+ const unsigned len = sizeof (buf);
+ unsigned pos = 0;
+ Slrn_Artnum_Entry_Type *entry = artnums->first;
+ buf[0] = 0;
+
+ while (entry != NULL)
+ {
+ if (pos >= len - 1)
+ abort ();
+ snprintf (buf + pos, len - pos, "XOVER " NNTP_FMT_ARTRANGE "\r\n", entry->min, entry->max);
+ pos += strlen (buf + pos);
+ entry = entry->next;
+ }
+
+ write (2, buf, pos);
+ nntp_server_pipelined_cmds (s, buf, pos);
+ return 0;
}
int nntp_xhdr_cmd (NNTP_Type *s, char *field, NNTP_Artnum_Type min, NNTP_Artnum_Type max)
diff --git a/src/nntplib.h b/src/nntplib.h
index 14ae57c..50c2e77 100644
--- a/src/nntplib.h
+++ b/src/nntplib.h
@@ -68,6 +68,7 @@ extern int nntp_start_server_cmd (NNTP_Type *, char *);
extern int nntp_start_server_vcmd (NNTP_Type *, char *, ...) ATTRIBUTE_PRINTF(2,3);
extern int nntp_server_cmd (NNTP_Type *, char *);
extern int nntp_server_vcmd (NNTP_Type *, char *, ...) ATTRIBUTE_PRINTF(2,3);
+extern int nntp_server_pipelined_cmds (NNTP_Type *, const char *, unsigned);
extern char *nntp_get_server_name (void);
@@ -93,7 +94,7 @@ extern int nntp_one_xhdr_cmd (NNTP_Type *, char *, NNTP_Artnum_Type, char *, uns
extern int nntp_listgroup (NNTP_Type *, char *);
extern int nntp_head_cmd (NNTP_Type *, NNTP_Artnum_Type, char *, NNTP_Artnum_Type *);
-extern int nntp_xover_cmd (NNTP_Type *, NNTP_Artnum_Type, NNTP_Artnum_Type);
+extern int nntp_xover_cmd (NNTP_Type *, const Slrn_Artnum_List_Type *);
extern int nntp_xhdr_cmd (NNTP_Type *, char *, NNTP_Artnum_Type, NNTP_Artnum_Type);
extern int nntp_next_cmd (NNTP_Type *s, NNTP_Artnum_Type *);
extern int nntp_body_cmd (NNTP_Type *s, NNTP_Artnum_Type, char *);
diff --git a/src/server.c b/src/server.c
index bf860d5..f929c24 100644
--- a/src/server.c
+++ b/src/server.c
@@ -691,3 +691,48 @@ static int pull_parse_args (char **argv, int argc)
}
#endif
+
+void slrn_artnum_list_init (Slrn_Artnum_List_Type *list)
+{
+ list->first = NULL;
+ list->last = NULL;
+}
+
+void slrn_artnum_list_clear (Slrn_Artnum_List_Type *list)
+{
+ Slrn_Artnum_Entry_Type *entry = list->first, *next;
+ while (entry != NULL) {
+ next = entry->next;
+ free (entry);
+ entry = next;
+ }
+
+ list->first = NULL;
+ list->last = NULL;
+}
+
+void slrn_artnum_list_append (Slrn_Artnum_List_Type *list, NNTP_Artnum_Type min, NNTP_Artnum_Type max)
+{
+ Slrn_Artnum_Entry_Type *entry = malloc (sizeof (*entry));
+ entry->min = min;
+ entry->max = max;
+ entry->next = NULL;
+
+ if (list->last != NULL)
+ list->last->next = entry;
+ else
+ list->first = entry;
+ list->last = entry;
+}
+
+unsigned slrn_artnum_list_len (Slrn_Artnum_List_Type *list)
+{
+ unsigned r = 0;
+ Slrn_Artnum_Entry_Type *elem = list->first;
+ while (elem != NULL)
+ {
+ r++;
+ elem = elem->next;
+ }
+ return r;
+}
diff --git a/src/server.h b/src/server.h
index b77717c..f5d0d92 100644
--- a/src/server.h
+++ b/src/server.h
@@ -23,6 +23,19 @@
#include "nntpcodes.h"
#include "ranges.h"
+typedef struct Slrn_Artnum_Entry_Type
+{
+ NNTP_Artnum_Type min, max;
+ struct Slrn_Artnum_Entry_Type *next;
+}
+Slrn_Artnum_Entry_Type;
+
+typedef struct
+{
+ Slrn_Artnum_Entry_Type *first, *last;
+}
+Slrn_Artnum_List_Type;
+
typedef struct
{
int (*po_start)(void);
@@ -66,7 +79,8 @@ typedef struct
* This is because some servers support XOVER but do not have overview
* files for all groups. See xover.c
*/
- int (*sv_nntp_xover) (NNTP_Artnum_Type, NNTP_Artnum_Type);
+ int (*sv_nntp_send_xover) (const Slrn_Artnum_List_Type *);
+ int (*sv_nntp_xover_reply) (int min, int max);
int (*sv_nntp_xhdr) (char *, NNTP_Artnum_Type, NNTP_Artnum_Type);
int (*sv_nntp_head) (NNTP_Artnum_Type, char *, NNTP_Artnum_Type *);
int (*sv_nntp_next) (NNTP_Artnum_Type *);
@@ -83,6 +97,11 @@ typedef struct
}
Slrn_Server_Obj_Type;
+void slrn_artnum_list_init (Slrn_Artnum_List_Type *list);
+void slrn_artnum_list_clear (Slrn_Artnum_List_Type *list);
+void slrn_artnum_list_append (Slrn_Artnum_List_Type *list, NNTP_Artnum_Type min, NNTP_Artnum_Type max);
+unsigned slrn_artnum_list_len (Slrn_Artnum_List_Type *list);
+
extern Slrn_Server_Obj_Type *Slrn_Server_Obj;
extern Slrn_Post_Obj_Type *Slrn_Post_Obj;
diff --git a/src/sltcp.c b/src/sltcp.c
index a3042d3..33c0c9c 100644
--- a/src/sltcp.c
+++ b/src/sltcp.c
@@ -141,6 +141,8 @@ struct _SLTCP_Type
unsigned char tcp_write_buf [SLTCP_BUF_SIZE];
unsigned long bytes_out;
unsigned long bytes_in;
+ const char *write_buf;
+ unsigned long write_pending;
#if SLTCP_HAS_SSL_SUPPORT
SSL *ssl;
#endif
@@ -884,7 +886,7 @@ int sltcp_close (SLTCP_Type *tcp) /*{{{*/
#if !defined(VMS) && !defined(USE_WINSOCK_SLTCP)
static int wait_for_input (SLTCP_Type *tcp) /*{{{*/
{
- fd_set fds;
+ fd_set readfds, writefds;
struct timeval tv;
int fd = tcp->tcp_fd;
@@ -900,15 +902,32 @@ static int wait_for_input (SLTCP_Type *tcp) /*{{{*/
{
int ret;
- FD_ZERO(&fds);
- FD_SET(fd, &fds);
+ FD_ZERO(&readfds);
+ FD_ZERO(&writefds);
+ FD_SET(fd, &readfds);
+ if (tcp->write_pending != 0)
+ FD_SET(fd, &writefds);
tv.tv_sec = SLtcp_TimeOut_Secs;
tv.tv_usec = 0;
- ret = select(fd + 1, &fds, NULL, NULL, &tv);
+ ret = select(fd + 1, &readfds, &writefds, NULL, &tv);
if (ret > 0)
- return 0;
+ {
+ if (FD_ISSET(fd, &writefds))
+ {
+ int r = write (fd, tcp->write_buf, tcp->write_pending);
+ if (r <= 0 || r > tcp->write_pending)
+ abort ();
+ tcp->write_buf += r;
+ tcp->write_pending -= r;
+ }
+
+ if (FD_ISSET(fd, &readfds))
+ return 0;
+
+ continue;
+ }
if (ret == 0)
return -1; /* timed out */
@@ -1157,6 +1176,12 @@ int sltcp_fgets (SLTCP_Type *tcp, char *buf, unsigned int len) /*{{{*/
/*}}}*/
+int sltcp_queue_for_write (SLTCP_Type *tcp, const char *input, unsigned input_len)
+{
+ tcp->write_buf = input;
+ tcp->write_pending = input_len;
+}
+
/* Before any of the above routines in this file may be used, sltcp_open_sltcp
* must be called.
*/
diff --git a/src/sltcp.h b/src/sltcp.h
index 1ab4340..84e9dba 100644
--- a/src/sltcp.h
+++ b/src/sltcp.h
@@ -37,6 +37,8 @@ extern int sltcp_fputs (SLTCP_Type *, char *);
extern int sltcp_vfprintf (SLTCP_Type *, char *, va_list);
extern int sltcp_fgets (SLTCP_Type *, char *, unsigned int);
+extern int sltcp_queue_for_write (SLTCP_Type *, const char *, unsigned);
+
extern int sltcp_open_sltcp (void);
extern int sltcp_close_sltcp (void);
extern void (*SLtcp_Verror_Hook) (char *, va_list);
diff --git a/src/xover.c b/src/xover.c
index 34efe7d..8bbcafb 100644
--- a/src/xover.c
+++ b/src/xover.c
@@ -775,7 +775,12 @@ static int read_head_into_xover (NNTP_Artnum_Type id, Slrn_XOver_Type *xov) /*{{
extern int Slrn_Prefer_Head;
-int slrn_open_xover (NNTP_Artnum_Type min, NNTP_Artnum_Type max) /*{{{*/
+int slrn_send_xover (const Slrn_Artnum_List_Type *artnums)
+{
+ Slrn_Server_Obj->sv_nntp_send_xover (artnums);
+}
+
+int slrn_open_xover_reply (NNTP_Artnum_Type min, NNTP_Artnum_Type max) /*{{{*/
{
NNTP_Artnum_Type id;
int status = -1;
@@ -785,7 +790,7 @@ int slrn_open_xover (NNTP_Artnum_Type min, NNTP_Artnum_Type max) /*{{{*/
(Slrn_Prefer_Head != 2))
{
rearrange_add_headers ();
- status = Slrn_Server_Obj->sv_nntp_xover (min, max);
+ status = Slrn_Server_Obj->sv_nntp_xover_reply (min, max);
if (status == OK_XOVER)
{
XOver_Next = XOver_Min = min;
@@ -807,6 +812,8 @@ int slrn_open_xover (NNTP_Artnum_Type min, NNTP_Artnum_Type max) /*{{{*/
* article in this range. If the range is large an no articles are present
* in the range, it may be rather slow.
*/
+ /* XXX */
+#if 0
for (id = min; id <= max; id++)
{
status = (*Slrn_Server_Obj->sv_nntp_head)(id, NULL, NULL);
@@ -820,6 +827,7 @@ int slrn_open_xover (NNTP_Artnum_Type min, NNTP_Artnum_Type max) /*{{{*/
if (id > max)
return status;
+#endif
XOver_Next = XOver_Min = id;
XOver_Max = max;
@@ -1092,7 +1100,8 @@ int slrn_open_add_xover (NNTP_Artnum_Type min, NNTP_Artnum_Type max) /*{{{*/
/* Server does not support xhdr -- we need to get full HEADers (slow!) */
Suspend_XOver_For_Kill |= 2; /* quick hack */
- status = slrn_open_xover (min, max);
+ // XXX status = slrn_open_xover (min, max);
+ status = ERR_XFERFAIL;
return (status == OK_XOVER) ? 1 : -1;
}
diff --git a/src/xover.h b/src/xover.h
index 3515228..499ba50 100644
--- a/src/xover.h
+++ b/src/xover.h
@@ -21,6 +21,8 @@
#ifndef _SLRN_XOVER_H
#define _SLRN_XOVER_H
+#include "server.h"
+
/* In this structure, only subject_malloced will be malloced. All other
* pointers point to a location in that space. It is done this way because
* art.c uses this convention and the pointer can just be passed to it.
@@ -52,7 +54,8 @@ extern int slrn_read_overview_fmt (void);
#ifndef SLRNPULL_CODE
extern int slrn_xover_for_msgid (char *, Slrn_XOver_Type *);
-extern int slrn_open_xover (NNTP_Artnum_Type, NNTP_Artnum_Type);
+extern int slrn_send_xover (const Slrn_Artnum_List_Type *);
+extern int slrn_open_xover_reply (NNTP_Artnum_Type, NNTP_Artnum_Type);
extern int slrn_read_xover (Slrn_XOver_Type *);
extern void slrn_close_xover (void);
--
Jan Andres <[email protected]>
------------------------------------------------------------------------------
Live Security Virtual Conference
Exclusive live event will cover all the ways today's security and
threat landscape has changed and how IT managers can respond. Discussions
will include endpoint security, mobile security and the latest in malware
threats. http://www.accelacomm.com/jaw/sfrnl04242012/114/50122263/