Fsync changes.

OVERLAPPED structures.

CHANGELOG.md

@@ -2,6 +2,7 @@
 All notable changes to this project will be documented in this file.
 
 ## Unreleased - ???
+- Add `file/sync` as a wrapper around fsync.
 - Documentation fixes
 - ev/thread-chan deadlock bug fixed
 - Re-add removed support for non-blocking net/connect on windows.

build_win.bat

@@ -29,14 +29,16 @@ if DEFINED CLANG (
     @set COMPILER=cl.exe
 )
 if DEFINED SANITIZE (
-    @set "SANITIZERS=/fsanitize=address"
+    @set "SANITIZERS=/fsanitize=address /Zi"
+    @set "LINK_SAN=/DEBUG"
 ) else (
     @set "SANITIZERS="
+    @set "LINK_SAN=/DEBUG"
 )
 @set JANET_COMPILE=%COMPILER% /nologo /Isrc\include /Isrc\conf /c /O2 /W3 /D_CRT_SECURE_NO_WARNINGS /MD %SANITIZERS%
-@set JANET_LINK=link /nologo
+@set JANET_LINK=link /nologo %LINK_SAN%
 
-@set JANET_LINK_STATIC=lib /nologo
+@set JANET_LINK_STATIC=lib /nologo %LINK_SAN%
 
 @rem Add janet build tag
 if not "%JANET_BUILD%" == "" (

src/core/ev.c

@@ -968,7 +968,7 @@ static void janet_thread_chan_cb(JanetEVGenericMessage msg) {
             while (!janet_q_pop(&channel->read_pending, &reader, sizeof(reader))) {
                 JanetVM *vm = reader.thread;
                 if (!vm) continue;
-                JanetEVGenericMessage msg;
+                JanetEVGenericMessage msg = {0};
                 msg.tag = reader.mode;
                 msg.fiber = reader.fiber;
                 msg.argi = (int32_t) reader.sched_id;
@@ -986,7 +986,7 @@ static void janet_thread_chan_cb(JanetEVGenericMessage msg) {
             while (!janet_q_pop(&channel->write_pending, &writer, sizeof(writer))) {
                 JanetVM *vm = writer.thread;
                 if (!vm) continue;
-                JanetEVGenericMessage msg;
+                JanetEVGenericMessage msg = {0};
                 msg.tag = writer.mode;
                 msg.fiber = writer.fiber;
                 msg.argi = (int32_t) writer.sched_id;
@@ -1052,7 +1052,7 @@ static int janet_channel_push_with_lock(JanetChannel *channel, Janet x, int mode
         /* Pending reader */
         if (is_threaded) {
             JanetVM *vm = reader.thread;
-            JanetEVGenericMessage msg;
+            JanetEVGenericMessage msg = {0};
             msg.tag = reader.mode;
             msg.fiber = reader.fiber;
             msg.argi = (int32_t) reader.sched_id;
@@ -1112,7 +1112,7 @@ static int janet_channel_pop_with_lock(JanetChannel *channel, Janet *item, int i
         /* Pending writer */
         if (is_threaded) {
             JanetVM *vm = writer.thread;
-            JanetEVGenericMessage msg;
+            JanetEVGenericMessage msg = {0};
             msg.tag = writer.mode;
             msg.fiber = writer.fiber;
             msg.argi = (int32_t) writer.sched_id;
@@ -1364,7 +1364,7 @@ JANET_CORE_FN(cfun_channel_close,
         while (!janet_q_pop(&channel->write_pending, &writer, sizeof(writer))) {
             if (writer.thread != &janet_vm) {
                 JanetVM *vm = writer.thread;
-                JanetEVGenericMessage msg;
+                JanetEVGenericMessage msg = {0};
                 msg.fiber = writer.fiber;
                 msg.argp = channel;
                 msg.tag = JANET_CP_MODE_CLOSE;
@@ -1387,7 +1387,7 @@ JANET_CORE_FN(cfun_channel_close,
         while (!janet_q_pop(&channel->read_pending, &reader, sizeof(reader))) {
             if (reader.thread != &janet_vm) {
                 JanetVM *vm = reader.thread;
-                JanetEVGenericMessage msg;
+                JanetEVGenericMessage msg = {0};
                 msg.fiber = reader.fiber;
                 msg.argp = channel;
                 msg.tag = JANET_CP_MODE_CLOSE;
@@ -1722,7 +1722,7 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp to) {
             }
             if (fiber != NULL) {
                 fiber->flags &= ~JANET_FIBER_EV_FLAG_IN_FLIGHT;
-                jo->bytes_transfered = (ULONG_PTR) num_bytes_transferred;
+                jo->bytes_transferred = (ULONG_PTR) num_bytes_transferred;
                 fiber->ev_callback(fiber, result ? JANET_ASYNC_EVENT_COMPLETE : JANET_ASYNC_EVENT_FAILED);
             } else {
                 janet_free((void *) jo);
@@ -2257,11 +2257,14 @@ static DWORD WINAPI janet_thread_body(LPVOID ptr) {
     /* Reuse memory from thread init for returning data */
     init->msg = subr(msg);
     init->cb = cb;
-    janet_assert(PostQueuedCompletionStatus(iocp,
+    BOOL result = PostQueuedCompletionStatus(iocp,
                                             sizeof(JanetSelfPipeEvent),
                                             0,
-                                            (LPOVERLAPPED) init),
-                 "failed to post completion event");
+                                            (LPOVERLAPPED) init);
+    if (!result) {
+        JanetString x = janet_formatc("failed to post completion event: %V", janet_ev_lasterr());
+        janet_assert(0, (const char *)x);
+    }
     return 0;
 }
 #else
@@ -2363,8 +2366,7 @@ void janet_ev_default_threaded_callback(JanetEVGenericMessage return_value) {
 /* Convenience method for common case */
 JANET_NO_RETURN
 void janet_ev_threaded_await(JanetThreadedSubroutine fp, int tag, int argi, void *argp) {
-    JanetEVGenericMessage arguments;
-    memset(&arguments, 0, sizeof(arguments));
+    JanetEVGenericMessage arguments = {0};
     arguments.tag = tag;
     arguments.argi = argi;
     arguments.argp = argp;
@@ -2472,7 +2474,7 @@ void ev_callback_read(JanetFiber *fiber, JanetAsyncEvent event) {
         case JANET_ASYNC_EVENT_FAILED:
         case JANET_ASYNC_EVENT_COMPLETE: {
             /* Called when read finished */
-            uint32_t ev_bytes = (uint32_t) state->overlapped.bytes_transfered;
+            uint32_t ev_bytes = (uint32_t) state->overlapped.bytes_transferred;
             state->bytes_read += ev_bytes;
             if (state->bytes_read == 0 && (state->mode != JANET_ASYNC_READMODE_RECVFROM)) {
                 janet_schedule(fiber, janet_wrap_nil());
@@ -2722,7 +2724,7 @@ void ev_callback_write(JanetFiber *fiber, JanetAsyncEvent event) {
         case JANET_ASYNC_EVENT_FAILED:
         case JANET_ASYNC_EVENT_COMPLETE: {
             /* Called when write finished */
-            uint32_t ev_bytes = (uint32_t) state->overlapped.bytes_transfered;
+            uint32_t ev_bytes = (uint32_t) state->overlapped.bytes_transferred;
             if (ev_bytes == 0 && (state->mode != JANET_ASYNC_WRITEMODE_SENDTO)) {
                 janet_cancel(fiber, janet_cstringv("disconnect"));
                 janet_async_end(fiber);
@@ -3206,8 +3208,7 @@ JANET_CORE_FN(cfun_ev_thread,
     janet_marshal(buffer, value, NULL, JANET_MARSHAL_UNSAFE);
     if (flags & 0x1) {
         /* Return immediately */
-        JanetEVGenericMessage arguments;
-        memset(&arguments, 0, sizeof(arguments));
+        JanetEVGenericMessage arguments = {0};
         arguments.tag = (uint32_t) flags;
         arguments.argi = (uint32_t) janet_vm.sandbox_flags;
         arguments.argp = buffer;

src/core/io.c

@@ -320,6 +320,41 @@ static int cfun_io_gc(void *p, size_t len) {
     return 0;
 }
 
+/* Cross-platform fsync binding for Janet */
+JANET_CORE_FN(cfun_io_fsync,
+              "(file/sync f)",
+              "Flushes all operating system buffers to disk for file `f`. Guarantees data is physically "
+              "written to disk in a platform-dependent way. Returns the file handle if successful, raises error if not.") {
+    janet_fixarity(argc, 1);
+    JanetFile *iof = janet_getabstract(argv, 0, &janet_file_type);
+    if (iof->flags & JANET_FILE_CLOSED)
+        janet_panic("file is closed");
+#ifdef JANET_WINDOWS
+    {
+        int fd = _fileno(iof->file);
+        if (fd < 0)
+            janet_panic("invalid file descriptor");
+        HANDLE hFile = (HANDLE)_get_osfhandle(fd);
+        if (hFile == INVALID_HANDLE_VALUE)
+            janet_panic("invalid file handle");
+        if (!FlushFileBuffers(hFile))
+            janet_panic("could not flush file buffers");
+    }
+#elif defined(_POSIX_VERSION)
+    {
+        int fd = fileno(iof->file);
+        if (fd < 0)
+            janet_panic("invalid file descriptor");
+        if (fsync(fd) != 0)
+            janet_panic("could not fsync file");
+    }
+#else
+    janet_panic("fsync not supported on this platform");
+#endif
+    return argv[0];
+}
+
+
 /* Close a file */
 JANET_CORE_FN(cfun_io_fclose,
               "(file/close f)",
@@ -394,6 +429,7 @@ static JanetMethod io_file_methods[] = {
     {"seek", cfun_io_fseek},
     {"tell", cfun_io_ftell},
     {"write", cfun_io_fwrite},
+    {"sync", cfun_io_fsync},
     {NULL, NULL}
 };
 
@@ -846,6 +882,7 @@ void janet_lib_io(JanetTable *env) {
         JANET_CORE_REG("file/flush", cfun_io_fflush),
         JANET_CORE_REG("file/seek", cfun_io_fseek),
         JANET_CORE_REG("file/tell", cfun_io_ftell),
+        JANET_CORE_REG("file/sync", cfun_io_fsync),
         JANET_REG_END
     };
     janet_core_cfuns_ext(env, NULL, io_cfuns);

src/core/util.h

@@ -213,7 +213,7 @@ typedef struct {
         OVERLAPPED overlapped;
         WSAOVERLAPPED wsaoverlapped;
     } as;
-    uint32_t bytes_transfered;
+    uint32_t bytes_transferred;
 } JanetOverlapped;
 #endif
 #endif

src/include/janet.h

@@ -1531,6 +1531,9 @@ JANET_API Janet janet_ev_lasterr(void);
  * We could just use a pointer but this prevents malloc/free in the common case
  * of only a handful of arguments. */
 typedef struct {
+#ifdef JANET_WINDOWS
+    char padding[48]; /* On windows, used for OVERLAPPED storage */
+#endif
     int tag;
     int argi;
     void *argp;

test/suite-io.janet

@@ -55,7 +55,8 @@
   (file/flush f)
   (file/seek f :set 0)
   (assert (= 0 (file/tell f)) "start of file again")
-  (assert (= (string (file/read f :all)) "foo\n") "temp files work"))
+  (assert (= (string (file/read f :all)) "foo\n") "temp files work")
+  (assert-no-error "fsync" (file/sync f)))
 
 # issue #1055 - 2c927ea76
 (let [b @""]
@@ -74,9 +75,13 @@
 (defn to-b [a] (buffer/push b a))
 (xprintf to-b "123")
 (assert (deep= b @"123\n") "xprintf to buffer")
-
-
 (assert-error "cannot print to 3" (xprintf 3 "123"))
 
+# file/sync
+(with [f (file/temp)]
+  (file/write f "123abc")
+  (file/flush f)
+  (file/sync f))
+
 (end-suite)