Teach prim_file:list_dir/1,2 to use fname chunks

The efile driver will now use chunked data on list_dir. This will
lessen the number of sends to prim_file and hence improve
performance. This method is utilized in both direct and async cases.

2 files changed, 99 insertions, 71 deletions

diff --git a/erts/emulator/drivers/common/efile_drv.c b/erts/emulator/drivers/common/efile_drv.c
index f0ff3f54c5..3c9007311f 100644
--- a/erts/emulator/drivers/common/efile_drv.c
+++ b/erts/emulator/drivers/common/efile_drv.c
@@ -69,6 +69,7 @@
 #define FILE_RESP_EOF        8
 #define FILE_RESP_FNAME      9
 #define FILE_RESP_ALL_DATA  10
+#define FILE_RESP_LFNAME    11
 
 /* Options */
 
@@ -184,6 +185,7 @@ static ErlDrvSysInfo sys_info;
 #  define    RESBUFSIZE  BUFSIZ
 #endif
 
+#define READDIR_CHUNKS (5)
 
 
 
@@ -317,15 +319,16 @@ struct t_preadv {
     Sint64   offsets[1];
 };
 
-#define READDIR_BUFSIZE (8*1024)
-#if READDIR_BUFSIZE < (FILENAME_CHARSIZE*2*(MAXPATHLEN+1))
+#define READDIR_BUFSIZE (8*1024)*READDIR_CHUNKS
+#if READDIR_BUFSIZE < (1 + (2 + MAXPATHLEN)*FILENAME_CHARSIZE*READDIR_CHUNKS)
 #  undef READDIR_BUFSIZE
-#  define READDIR_BUFSIZE (FILENAME_CHARSIZE*2*(MAXPATHLEN+1))
+#  define READDIR_BUFSIZE (1 + (2 + MAXPATHLEN)*FILENAME_CHARSIZE*READDIR_CHUNKS)
 #endif
 
 struct t_readdir_buf {
-    struct t_readdir_buf *next;
-    char buf[READDIR_BUFSIZE];
+     struct t_readdir_buf *next;
+     size_t n;
+     char buf[READDIR_BUFSIZE];
 };
 
 struct t_data
@@ -1598,54 +1601,43 @@ static void invoke_lseek(void *data)
 static void invoke_readdir(void *data)
 {
     struct t_data *d = (struct t_data *) data;
-    int s;
     char *p = NULL;
-    int buf_sz = 0;
-    size_t tmp_bs;
+    size_t file_bs;
+    size_t n = 0, total = 0;
+    struct t_readdir_buf *b = NULL;
+    int res = 0;
 
     d->again = 0;
     d->errInfo.posix_errno = 0;
 
-    while (1) {
-	char *str;
-	if (buf_sz < (4 /* sz */ + 1 /* cmd */ + 
-		      FILENAME_CHARSIZE*(MAXPATHLEN + 1))) {
-	    struct t_readdir_buf *b;
-	    if (p) {
-		put_int32(0, p); /* EOB */
-	    }
-	    b = EF_SAFE_ALLOC(sizeof(struct t_readdir_buf));
-	    b->next = NULL;
-	    if (d->c.read_dir.last_buf)
-		d->c.read_dir.last_buf->next = b;
-	    else
-		d->c.read_dir.first_buf = b;
-	    d->c.read_dir.last_buf = b;
-	    p = &b->buf[0];
-	    buf_sz = READDIR_BUFSIZE - 4/* EOB */;
-	}
-
-	p[4] = FILE_RESP_FNAME;
-	buf_sz -= 4 + 1;
-	str = p + 4 + 1;
-	ASSERT(buf_sz >= MAXPATHLEN + 1);
-	tmp_bs = buf_sz;
-	s = efile_readdir(&d->errInfo, d->b, &d->dir_handle, str, &tmp_bs);
-
-	if (s) {
-	    put_int32(tmp_bs + 1 /* 1 byte for opcode */, p);
-	    p += 4 + tmp_bs + 1;
-	    ASSERT(p == (str + tmp_bs));
-	    buf_sz -= tmp_bs;
+    do {
+	total   = READDIR_BUFSIZE;
+	n       = 1;
+	b       = EF_SAFE_ALLOC(sizeof(struct t_readdir_buf));
+	b->next = NULL;
+	
+	if (d->c.read_dir.last_buf) {
+	    d->c.read_dir.last_buf->next = b;
+	} else {
+	    d->c.read_dir.first_buf = b;
 	}
-	else {
-	    put_int32(1, p);
-	    p += 4 + 1;
-	    put_int32(0, p); /* EOB */
-	    d->result_ok = (d->errInfo.posix_errno == 0);
-	    break;
+	d->c.read_dir.last_buf = b;
+
+	p       = &b->buf[0];
+	p[0]    = FILE_RESP_LFNAME;
+	file_bs = READDIR_BUFSIZE - n;
+
+	while ( (res = efile_readdir(&d->errInfo, d->b, &d->dir_handle, p + n + 2, &file_bs)) 
+		&& ((total - n - 2) >= MAXPATHLEN*FILENAME_CHARSIZE)) {
+	    put_int16((Uint16)file_bs, p + n);
+	    n += 2 + file_bs;
+	    file_bs = READDIR_BUFSIZE - n;
 	}
-    }
+
+	b->n = n;
+    } while(res);
+
+    d->result_ok = (d->errInfo.posix_errno == 0);
 }
 
 static void invoke_open(void *data)
@@ -2053,30 +2045,24 @@ file_async_ready(ErlDrvData e, ErlDrvThreadData data)
 	free_data(data);
 	break;
       case FILE_READDIR:
-	if (!d->result_ok)
+	if (!d->result_ok) {
 	    reply_error(desc, &d->errInfo);
-	else {
+	} else {
 	    struct t_readdir_buf *b1 = d->c.read_dir.first_buf;
+	    char   op = FILE_RESP_LFNAME;
+
 	    TRACE_C('R');
 	    ASSERT(b1);
+
 	    while (b1) {
 		struct t_readdir_buf *b2 = b1;
 		char *p = &b1->buf[0];
-		int sz = get_int32(p);
-		while (sz) { /* 0 == EOB */
-		    p += 4;
-		    if (sz - 1 > 0) {
-			driver_output2(desc->port, p, 1, p+1, sz-1);
-		    } else {
-			driver_output2(desc->port, p, 1, NULL, 0);
-		    }
-		    p += sz;
-		    sz = get_int32(p);
-		}
+		driver_output2(desc->port, p, 1, p + 1, b1->n - 1);
 		b1 = b1->next;
 		EF_FREE(b2);
 	    }
-	    
+	    driver_output2(desc->port, &op, 1, NULL, 0);
+    
 	    d->c.read_dir.first_buf = NULL;
 	    d->c.read_dir.last_buf = NULL;
 	}
@@ -2246,19 +2232,42 @@ file_output(ErlDrvData e, char* buf, int count)
 #endif
 	{
 	    size_t resbufsize;
-	    char resbuf[RESBUFSIZE+1];
+	    size_t n = 0, total = 0;
+	    int res = 0;
+	    char resbuf[READDIR_BUFSIZE];
+
 	    EFILE_DIR_HANDLE dir_handle; /* Handle to open directory. */
 
+	    total               = READDIR_BUFSIZE;
 	    errInfo.posix_errno = 0;
-	    dir_handle = NULL;
-	    resbuf[0] = FILE_RESP_FNAME;
-	    resbufsize = RESBUFSIZE;
-
-	    while (efile_readdir(&errInfo, name, &dir_handle,
-				 resbuf+1, &resbufsize)) {
-		driver_output2(desc->port, resbuf, 1, resbuf+1, resbufsize);
-		resbufsize = RESBUFSIZE;
-	    }
+	    dir_handle          = NULL;
+	    resbuf[0]           = FILE_RESP_LFNAME;
+
+	    /* Fill the buffer with multiple directory listings before sending it to the
+	     * receiving process. READDIR_CHUNKS is minimum number of files sent to the
+	     * receiver.
+	     * Format for each driver_output2:
+	     * ------------------------------------
+	     * | Type   | Len     | Filename  | ...
+	     * | 1 byte | 2 bytes | Len bytes | ...
+	     * ------------------------------------
+	     */
+
+	    do {
+		n = 1;
+		resbufsize = READDIR_BUFSIZE - n;
+		while ( (res = efile_readdir(&errInfo, name, &dir_handle, resbuf + n + 2, &resbufsize)) 
+			&& ((total - n - 2) >= MAXPATHLEN*FILENAME_CHARSIZE)) {
+		    put_int16((Uint16)resbufsize, resbuf + n);
+		    n += 2 + resbufsize;
+		    resbufsize = READDIR_BUFSIZE - n;
+		}
+
+		if (n > 1) {
+		    driver_output2(desc->port, resbuf, 1, resbuf+1, n - 1);
+		}
+	    } while(res);
+
 	    if (errInfo.posix_errno != 0) {
 		reply_error(desc, &errInfo);
 		return;
diff --git a/erts/preloaded/src/prim_file.erl b/erts/preloaded/src/prim_file.erl
index ac7570582e..b18a57bfef 100644
--- a/erts/preloaded/src/prim_file.erl
+++ b/erts/preloaded/src/prim_file.erl
@@ -111,6 +111,7 @@
 -define(FILE_RESP_EOF,         8).
 -define(FILE_RESP_FNAME,       9).
 -define(FILE_RESP_ALL_DATA,   10).
+-define(FILE_RESP_LFNAME,     11).
 
 %% Open modes for the driver's open function.
 -define(EFILE_MODE_READ,       1).
@@ -914,6 +915,8 @@ drv_get_response(Port, R) when is_list(R) ->
 	    {ok, R};
 	{ok, Name} ->
 	    drv_get_response(Port, [Name|R]);
+	{append, Names} ->
+	    drv_get_response(Port, append(Names, R));
 	Error ->
 	    Error
     end;
@@ -938,6 +941,8 @@ drv_get_response(Port) ->
 %%%-----------------------------------------------------------------
 %%% Utility functions.
 
+append([I | Is], R) when is_list(R) -> append(Is, [I | R]);
+append([], R) -> R.
 
 
 %% Converts a list of mode atoms into an mode word for the driver.
@@ -1067,7 +1072,7 @@ translate_response(?FILE_RESP_N2DATA,
     {ok, {Size, Offset, D}};
 translate_response(?FILE_RESP_N2DATA = X, L0) when is_list(L0) ->
     {Offset, L1}    = get_uint64(L0),
-    {ReadSize, L2} = get_uint64(L1),
+    {ReadSize, L2}  = get_uint64(L1),
     {Size, L3}      = get_uint64(L2),
     case {ReadSize, L3} of
 	{0, []} ->
@@ -1087,6 +1092,12 @@ translate_response(?FILE_RESP_FNAME, Data) when is_binary(Data) ->
     {ok, prim_file:internal_native2name(Data)};
 translate_response(?FILE_RESP_FNAME, Data) ->
     {ok, Data};
+translate_response(?FILE_RESP_LFNAME, []) ->
+    ok;
+translate_response(?FILE_RESP_LFNAME, Data) when is_binary(Data) ->
+    {append, transform_lfname(Data)};
+translate_response(?FILE_RESP_LFNAME, Data) ->
+    {append, transform_lfname(Data)};
 translate_response(?FILE_RESP_ALL_DATA, Data) ->
     {ok, Data};
 translate_response(X, Data) ->
@@ -1199,6 +1210,14 @@ transform_ldata(0, List, [Size | Sizes], R) ->
     {Front, Rear} = lists_split(List, Size),
     transform_ldata(0, Rear, Sizes, [Front | R]).
 
+transform_lfname(<<>>) -> [];
+transform_lfname(<<L:16, Name:L/binary, Names/binary>>) -> 
+    [ prim_file:internal_native2name(Name) | transform_lfname(Names)];
+transform_lfname([]) -> [];
+transform_lfname([L1,L2|Names]) ->
+    L = (L1 bsl 8) bor L2,
+    {Name, Rest} = lists_split(Names, L),
+    [Name | transform_lfname(Rest)].
 
 
 lists_split(List, 0) when is_list(List) ->