use O_APPEND to skip some lseeks, COUCHDB-754

git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@945110 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 50 insertions, 17 deletions

diff --git a/src/couchdb/couch_file.erl b/src/couchdb/couch_file.erl
index 7a4c32a3..0d79afb2 100644
--- a/src/couchdb/couch_file.erl
+++ b/src/couchdb/couch_file.erl
@@ -19,7 +19,8 @@
 
 -record(file, {
     fd,
-    tail_append_begin=0 % 09 UPGRADE CODE
+    tail_append_begin = 0, % 09 UPGRADE CODE
+    eof = 0
     }).
 
 -export([open/1, open/2, close/1, bytes/1, sync/1, append_binary/2,old_pread/3]).
@@ -204,7 +205,7 @@ init({Filepath, Options, ReturnPid, Ref}) ->
     case lists:member(create, Options) of
     true ->
         filelib:ensure_dir(Filepath),
-        case file:open(Filepath, [read, write, raw, binary]) of
+        case file:open(Filepath, [read, append, raw, binary]) of
         {ok, Fd} ->
             {ok, Length} = file:position(Fd, eof),
             case Length > 0 of
@@ -236,10 +237,11 @@ init({Filepath, Options, ReturnPid, Ref}) ->
         % open in read mode first, so we don't create the file if it doesn't exist.
         case file:open(Filepath, [read, raw]) of
         {ok, Fd_Read} ->
-            {ok, Fd} = file:open(Filepath, [read, write, raw, binary]),
+            {ok, Fd} = file:open(Filepath, [read, append, raw, binary]),
             ok = file:close(Fd_Read),
             couch_stats_collector:track_process_count({couchdb, open_os_files}),
-            {ok, #file{fd=Fd}};
+            {ok, Length} = file:position(Fd, eof),
+            {ok, #file{fd=Fd, eof=Length}};
         Error ->
             init_status_error(ReturnPid, Ref, Error)
         end
@@ -269,24 +271,27 @@ handle_call({pread_iolist, Pos}, _From, File) ->
 handle_call({pread, Pos, Bytes}, _From, #file{fd=Fd,tail_append_begin=TailAppendBegin}=File) ->
     {ok, Bin} = file:pread(Fd, Pos, Bytes),
     {reply, {ok, Bin, Pos >= TailAppendBegin}, File};
-handle_call(bytes, _From, #file{fd=Fd}=File) ->
-    {reply, file:position(Fd, eof), File};
+handle_call(bytes, _From, #file{eof=Length}=File) ->
+    {reply, {ok, Length}, File};
 handle_call(sync, _From, #file{fd=Fd}=File) ->
     {reply, file:sync(Fd), File};
 handle_call({truncate, Pos}, _From, #file{fd=Fd}=File) ->
     {ok, Pos} = file:position(Fd, Pos),
-    {reply, file:truncate(Fd), File};
-handle_call({append_bin, Bin}, _From, #file{fd=Fd}=File) ->
-    {ok, Pos} = file:position(Fd, eof),
+    case file:truncate(Fd) of
+    ok ->
+        {reply, ok, File#file{eof=Pos}};
+    Error ->
+        {reply, Error, File}
+    end;
+handle_call({append_bin, Bin}, _From, #file{fd=Fd, eof=Pos}=File) ->
     Blocks = make_blocks(Pos rem ?SIZE_BLOCK, Bin),
-    case file:pwrite(Fd, Pos, Blocks) of
+    case file:write(Fd, Blocks) of
     ok ->
-        {reply, {ok, Pos}, File};
+        {reply, {ok, Pos}, File#file{eof=Pos+iolist_size(Blocks)}};
     Error ->
         {reply, Error, File}
     end;
-handle_call({write_header, Bin}, _From, #file{fd=Fd}=File) ->
-    {ok, Pos} = file:position(Fd, eof),
+handle_call({write_header, Bin}, _From, #file{fd=Fd, eof=Pos}=File) ->
     BinSize = size(Bin),
     case Pos rem ?SIZE_BLOCK of
     0 ->
@@ -295,13 +300,18 @@ handle_call({write_header, Bin}, _From, #file{fd=Fd}=File) ->
         Padding = <<0:(8*(?SIZE_BLOCK-BlockOffset))>>
     end,
     FinalBin = [Padding, <<1, BinSize:32/integer>> | make_blocks(1, [Bin])],
-    {reply, file:pwrite(Fd, Pos, FinalBin), File};
+    case file:write(Fd, FinalBin) of
+    ok ->
+        {reply, ok, File#file{eof=Pos+iolist_size(FinalBin)}};
+    Error ->
+        {reply, Error, File}
+    end;
 
 
 handle_call({upgrade_old_header, Prefix}, _From, #file{fd=Fd}=File) ->
     case (catch read_old_header(Fd, Prefix)) of
     {ok, Header} ->
-        {ok, TailAppendBegin} = file:position(Fd, eof),
+        TailAppendBegin = File#file.eof,
         Bin = term_to_binary(Header),
         Md5 = couch_util:md5(Bin),
         % now we assemble the final header binary and write to disk
@@ -319,8 +329,7 @@ handle_call({upgrade_old_header, Prefix}, _From, #file{fd=Fd}=File) ->
     end;
 
 
-handle_call(find_header, _From, #file{fd=Fd}=File) ->
-    {ok, Pos} = file:position(Fd, eof),
+handle_call(find_header, _From, #file{fd=Fd, eof=Pos}=File) ->
     {reply, find_header(Fd, Pos div ?SIZE_BLOCK), File}.
 
 % 09 UPGRADE CODE
diff --git a/src/couchdb/priv/icu_driver/couch_icu_driver.c b/src/couchdb/priv/icu_driver/couch_icu_driver.c
index 1afe8eac..0bb843b5 100644
--- a/src/couchdb/priv/icu_driver/couch_icu_driver.c
+++ b/src/couchdb/priv/icu_driver/couch_icu_driver.c
@@ -27,8 +27,11 @@ specific language governing permissions and limitations under the License.
 #include "unicode/ucasemap.h"
 #ifndef WIN32
 #include <string.h> // for memcpy
+#include <fcntl.h> // for O_DSYNC
 #endif
 
+#define SET_OSYNC_OPCODE 2
+
 typedef struct {
     ErlDrvPort port;
     UCollator* collNoCase;
@@ -90,6 +93,19 @@ static int return_control_result(void* pLocalResult, int localLen, char **ppRetB
     return localLen;
 }
 
+static char set_osync(int fd) {
+    int flags = fcntl(fd, F_GETFL, 0);
+    if (flags != -1) {
+        if (fcntl(fd, F_SETFL, flags | O_DSYNC) != -1) {
+            return 0;
+        } else {
+            return 1;
+        }
+    } else {
+        return 2;
+    }
+}
+
 static int couch_drv_control(ErlDrvData drv_data, unsigned int command, char *pBuf,
              int bufLen, char **rbuf, int rlen)
 {
@@ -141,6 +157,14 @@ static int couch_drv_control(ErlDrvData drv_data, unsigned int command, char *pB
         return return_control_result(&response, sizeof(response), rbuf, rlen);
         }
 
+    case SET_OSYNC_OPCODE: // set O_SYNC flag on file descriptor
+    {
+        int32_t fd;
+        memcpy(&fd, pBuf, sizeof(fd));
+        char response = set_osync(fd);
+        return return_control_result(&response, sizeof(response), rbuf, rlen);
+    }
+
     default:
         return -1;
     }