Class: Winwatch::Watcher

Inherits:

Object

Object
Winwatch::Watcher

show all

Defined in:: lib/winwatch.rb,
ext/winwatch/winwatch.c

Overview

The directory watcher. Constructed only through Winwatch.watch (.new is private). The native superclass supplies the underscore-private primitives (_open / _take / _issue / _cancel / close* / closed? / _parse bridge).

Instance Attribute Summary collapse

#buffer_size ⇒ Object readonly
---- public accessors --------------------------------------------------.
#mode ⇒ Object readonly
---- public accessors --------------------------------------------------.
#path ⇒ Object readonly
---- public accessors --------------------------------------------------.

Class Method Summary collapse

._open(vpath, vfilter, vrec, vbuf, vext) ⇒ Object
Watcher._open(abs_path_utf8, filter_dword, recursive_bool, buffer_size, external_bool) -> Watcher.

Instance Method Summary collapse

#_arm ⇒ Object
Watcher#_arm -> Integer (0 = queued, else GetLastError).
#_cancel ⇒ Object
Watcher#_cancel — CancelIoEx(dir, NULL); safe from any thread; idempotent.
#_close_handle ⇒ Object
Watcher#_close_handle — settle + CloseHandle(s); idempotent; both modes.
#_close_standalone ⇒ Object
Watcher#_close_standalone — the full standalone close sequence (§3.5 rule 3): mark closed -> CancelIoEx(dir, &ov) -> deferred teardown tail iff in_take==0.
#_external? ⇒ Boolean
Watcher#external? — :winloop mode flag (set at _open).
#_handle_value ⇒ Object
Watcher#_handle_value -> Integer (the directory HANDLE, for sched.op_associate in :winloop mode).
#_in_take ⇒ Object
Watcher#in_take — diagnostic (used by the deferred-teardown stress test).
#_issue(vov, vbuf, vlen) ⇒ Object
Watcher#_issue(ov_addr, buf_addr, len) -> Integer (0 = queued, else GLE).
#_mark_closed ⇒ Object
Watcher#_mark_closed — set the closed flag (under the issue lock in Ruby for :winloop; standalone close uses _close which sets it directly).
#_take(vms) ⇒ Object
Watcher#_take(ms) -> nil | [data(String|nil), code(Int), rearm_code(Int)].
#close ⇒ Object
Cancel the in-flight op, close the directory handle, and wake any blocked take (which raises Closed).
#closed? ⇒ Boolean
Watcher#closed? — raw getter; works on closed objects.
#each ⇒ Object
Yield events one at a time, forever.
#filter ⇒ Object
#recursive? ⇒ Boolean
#take(timeout: nil) ⇒ Object
Returns a non-empty Array of Winwatch::Event as soon as any are available, or nil on timeout.

Instance Attribute Details

#buffer_size ⇒ `Object` (readonly)

---- public accessors --------------------------------------------------



204
205
206

# File 'lib/winwatch.rb', line 204

def buffer_size
  @buffer_size
end

#mode ⇒ `Object` (readonly)

---- public accessors --------------------------------------------------



204
205
206

# File 'lib/winwatch.rb', line 204

def mode
  @mode
end

#path ⇒ `Object` (readonly)

---- public accessors --------------------------------------------------



204
205
206

# File 'lib/winwatch.rb', line 204

def path
  @path
end

Class Method Details

._open(vpath, vfilter, vrec, vbuf, vext) ⇒ `Object`

Watcher._open(abs_path_utf8, filter_dword, recursive_bool, buffer_size, external_bool) -> Watcher. external_bool=true (:winloop mode) skips event/buffer creation and the first arm (the pump issues against winloop-owned memory). Captures GetLastError immediately, frees the wide path, then raises (the handle is stored into the struct before anything that can raise, so the free hook owns it).

# File 'ext/winwatch/winwatch.c', line 303

static VALUE
watch_open(VALUE klass, VALUE vpath, VALUE vfilter, VALUE vrec, VALUE vbuf, VALUE vext)
{
    VALUE obj = watch_alloc(cWatcher);
    watch_t *w = watch_get(obj);
    WCHAR *wpath;
    DWORD bufsz = NUM2ULONG(vbuf);
    DWORD gle;
    BY_HANDLE_FILE_INFORMATION fi;

    /* Range-assert again in C (§5.18). Round up to a DWORD multiple of 4. */
    if (bufsz < 4096 || bufsz > 65536)
        rb_raise(rb_eArgError, "winwatch: buffer_size out of range");
    bufsz = (bufsz + 3u) & ~3u;

    w->filter    = NUM2ULONG(vfilter);
    w->recursive = RTEST(vrec) ? TRUE : FALSE;
    w->external  = RTEST(vext) ? 1 : 0;
    w->buf_len   = bufsz;

    wpath = to_ext_wide(vpath); /* malloc'd; freed below before any raise */

    /* Share-all (READ|WRITE|DELETE) is load-bearing (§5.6): omitting
     * FILE_SHARE_DELETE blocks delete/rename of the root for other processes. */
    w->dir = CreateFileW(wpath, FILE_LIST_DIRECTORY,
                         FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
                         NULL, OPEN_EXISTING,
                         FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OVERLAPPED, NULL);
    gle = GetLastError();
    free(wpath);
    if (w->dir == INVALID_HANDLE_VALUE)
        raise_gle("CreateFile", gle); /* 2/3 NotFound, 5 AccessDenied, else OSError */

    /* It must be a directory (misuse, not OS). */
    if (!GetFileInformationByHandle(w->dir, &fi)) {
        gle = GetLastError();
        raise_gle("GetFileInformationByHandle", gle);
    }
    if (!(fi.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY))
        rb_raise(eNotADirectory, "winwatch: not a directory");

    if (!w->external) {
        w->ev = CreateEventW(NULL, TRUE, FALSE, NULL); /* manual-reset */
        if (!w->ev) raise_gle("CreateEvent", GetLastError());
        w->buf = (unsigned char *)malloc(w->buf_len);
        if (!w->buf) rb_raise(rb_eNoMemError, "winwatch: out of memory");

        /* First arm in the constructor: coverage starts before watch() returns
         * (rdcw §4/§14 re-arm discipline). A sync failure maps per §5.0. */
        gle = watch_arm(w);
        if (gle != 0)
            raise_gle("ReadDirectoryChangesW", gle); /* 1 Unsupported, 2/3 NotFound, 5 AccessDenied */
    }
    return obj;
}

Instance Method Details

#_arm ⇒ `Object`

Watcher#_arm -> Integer (0 = queued, else GetLastError). standalone only.

# File 'ext/winwatch/winwatch.c', line 360

static VALUE
watch_m_arm(VALUE self)
{
    watch_t *w = watch_live(self);
    return ULONG2NUM(watch_arm(w));
}

#_cancel ⇒ `Object`

Watcher#_cancel — CancelIoEx(dir, NULL); safe from any thread; idempotent.

# File 'ext/winwatch/winwatch.c', line 617

static VALUE
watch_cancel(VALUE self)
{
    watch_t *w = watch_get(self);
    if (w->dir != INVALID_HANDLE_VALUE)
        CancelIoEx(w->dir, NULL);
    return Qnil;
}

#_close_handle ⇒ `Object`

Watcher#_close_handle — settle + CloseHandle(s); idempotent; both modes. Only performs the teardown tail if no thread is inside _take (the in_take counter, §3.5 rule 3); otherwise the last leaver does it.

# File 'ext/winwatch/winwatch.c', line 640

static VALUE
watch_close_handle(VALUE self)
{
    watch_t *w = watch_get(self);
    if (w->in_take == 0)
        watch_teardown(w);
    return Qnil;
}

#_close_standalone ⇒ `Object`

Watcher#_close_standalone — the full standalone close sequence (§3.5 rule 3): mark closed -> CancelIoEx(dir, &ov) -> deferred teardown tail iff in_take==0. The cancel wakes any blocked _take (which re-checks closed first and returns the :closed indication; the last leaver performs the tail).

# File 'ext/winwatch/winwatch.c', line 655

static VALUE
watch_close_standalone(VALUE self)
{
    watch_t *w = watch_get(self);
    if (w->closed) return Qnil; /* idempotent */
    w->closed = 1;
    if (w->dir != INVALID_HANDLE_VALUE && w->armed)
        CancelIoEx(w->dir, &w->ov);
    if (w->in_take == 0)
        watch_teardown(w);
    /* else: the last leaver of _take runs the tail on its way out. */
    return Qnil;
}

#_external? ⇒ `Boolean`

Watcher#external? — :winloop mode flag (set at _open).

Returns:

(Boolean)

# File 'ext/winwatch/winwatch.c', line 685

static VALUE
watch_external_p(VALUE self)
{
    return watch_get(self)->external ? Qtrue : Qfalse;
}

#_handle_value ⇒ `Object`

Watcher#_handle_value -> Integer (the directory HANDLE, for sched.op_associate in :winloop mode). winwatch's struct stays the single owner of the handle.

# File 'ext/winwatch/winwatch.c', line 693

static VALUE
watch_handle_value(VALUE self)
{
    watch_t *w = watch_get(self);
    return ULL2NUM((unsigned long long)(uintptr_t)w->dir);
}

#_in_take ⇒ `Object`

Watcher#in_take — diagnostic (used by the deferred-teardown stress test).

# File 'ext/winwatch/winwatch.c', line 678

static VALUE
watch_in_take(VALUE self)
{
    return INT2NUM(watch_get(self)->in_take);
}

#_issue(vov, vbuf, vlen) ⇒ `Object`

Watcher#_issue(ov_addr, buf_addr, len) -> Integer (0 = queued, else GLE). :winloop only; calls RDCW with winloop's OVERLAPPED/buffer addresses.

# File 'ext/winwatch/winwatch.c', line 595

static VALUE
watch_issue(VALUE self, VALUE vov, VALUE vbuf, VALUE vlen)
{
    watch_t *w = watch_get(self);
    OVERLAPPED *ov = (OVERLAPPED *)(uintptr_t)NUM2ULL(vov);
    void *buf = (void *)(uintptr_t)NUM2ULL(vbuf);
    DWORD len = NUM2ULONG(vlen);
    BOOL ok;
    DWORD gle;

    if (w->dir == INVALID_HANDLE_VALUE)
        return ULONG2NUM(ERROR_OPERATION_ABORTED);

    ok = ReadDirectoryChangesW(w->dir, buf, len, w->recursive, w->filter,
                               NULL, ov, NULL);
    if (ok) return ULONG2NUM(0);
    gle = GetLastError();
    if (gle == ERROR_IO_PENDING) return ULONG2NUM(0);
    return ULONG2NUM(gle);
}

#_mark_closed ⇒ `Object`

Watcher#_mark_closed — set the closed flag (under the issue lock in Ruby for :winloop; standalone close uses _close which sets it directly). Returns the in_take count so the Ruby/close path could observe it if needed.

# File 'ext/winwatch/winwatch.c', line 629

static VALUE
watch_mark_closed(VALUE self)
{
    watch_t *w = watch_get(self);
    w->closed = 1;
    return INT2NUM(w->in_take);
}

#_take(vms) ⇒ `Object`

Watcher#_take(ms) -> nil | [data(String|nil), code(Int), rearm_code(Int)]. standalone only.

nil : timeout (no completion within ms)
[data, code, rc] : a completion (code = Win32 code, rc = re-arm result) data is a binary String (the raw batch) when bytes>0, else nil.
returns the symbol :closed via a distinct shape: we return Qnil only on timeout, and raise nothing here for close — the Ruby layer detects close by re-checking closed? after _take returns nil... To keep it unambiguous we instead signal close by returning the Ruby symbol :closed.

# File 'ext/winwatch/winwatch.c', line 408

static VALUE
watch_take(VALUE self, VALUE vms)
{
    watch_t *w = watch_live(self);
    /* Widen the conversion to 64-bit so a large-but-legitimate finite timeout
     * (ms > LONG_MAX on LLP64 where C long is 32-bit) does NOT raise RangeError
     * — the public API documents timeout: as non-negative Numeric seconds that
     * waits, and RangeError is not in the Raises list (§2.5). Negative => INFINITE
     * (the nil/forever spelling). Any value past the WaitForSingleObject finite
     * domain (0..0xFFFFFFFE; 0xFFFFFFFF is the INFINITE sentinel) clamps to the
     * max finite wait 0xFFFFFFFE (~49.7 days) rather than accidentally meaning
     * INFINITE. */
    long long ms_in = NUM2LL(vms);
    DWORD ms;
    take_wait_t t;
    DWORD bytes = 0, code = 0, rearm = 0;
    VALUE data = Qnil;
    BOOL got;

    if (ms_in < 0)
        ms = INFINITE;
    else if (ms_in > 0xFFFFFFFELL)
        ms = 0xFFFFFFFEu; /* clamp to the max finite WaitForSingleObject wait */
    else
        ms = (DWORD)ms_in;

    /* Serve a stashed batch first (interrupt-race losslessness, §5.12). */
    if (w->stash) {
        VALUE out;
        if (w->stash_len > 0) {
            data = rb_str_new((const char *)w->stash, (long)w->stash_len);
            rb_enc_associate(data, rb_ascii8bit_encoding());
        }
        code = w->stash_code;
        free(w->stash);
        w->stash = NULL;
        w->stash_len = 0;
        w->stash_code = 0;
        out = rb_ary_new_capa(3);
        rb_ary_push(out, data);
        rb_ary_push(out, ULONG2NUM(code));
        rb_ary_push(out, ULONG2NUM(0)); /* re-arm already done when we stashed */
        return out;
    }

    /* If no op is armed (e.g. a prior re-arm failed and Ruby asked us to wait),
     * try to re-arm right here and report the result as the rearm code, so a
     * successful re-arm restores the watch and a real failure is classified
     * (never an unarmed op that returns [nil,995,0] forever — the busy-livelock
     * fixed here). On a successful re-arm we still report a synthetic abort with
     * rearm==0 (no event, op re-armed, keep waiting); on a failed re-arm we
     * report the synthetic abort WITH the live rearm code so classify can turn a
     * permanent failure into a terminal :gone (or :rescan for 1022). */
    if (!w->armed) {
        DWORD rc = watch_arm(w);
        VALUE out = rb_ary_new_capa(3);
        rb_ary_push(out, Qnil);
        rb_ary_push(out, ULONG2NUM(ERROR_OPERATION_ABORTED));
        rb_ary_push(out, ULONG2NUM(rc));
        return out;
    }

    t.w = w; t.ms = ms; t.wait = WAIT_FAILED; t.gle = 0;

    w->in_take++;                       /* under the GVL, before release (§3.5 rule 3) */
    rb_thread_call_without_gvl(take_wait_fn, &t, take_wait_ubf, &t);
    w->in_take--;                       /* under the GVL, after reacquire */

    /* Closed-first check on every GVL reacquisition (§3.4). A concurrent close
     * ran: touch no dir/ov/buf. If we are the last leaver, perform the deferred
     * teardown tail (§3.5 rule 3). The Ruby layer raises Winwatch::Closed. */
    if (w->closed) {
        if (w->in_take == 0)
            watch_teardown(w);
        return ID2SYM(rb_intern("closed"));
    }

    if (t.wait == WAIT_TIMEOUT) {
        /* No completion within ms. But an interrupt (Thread#kill / Timeout)
         * may be pending — its ubf fired CancelIoEx, which can complete the op
         * with abort (or a real batch that lost the race). Settle the op into
         * the stash, re-arm, then deliver the interrupt. If no interrupt is
         * pending, the (abort) result is discarded and we return nil. */
        if (!w->armed) return Qnil; /* defensive */
        /* Probe without blocking: did the op complete (cancel or data)? */
        got = GetOverlappedResult(w->dir, &w->ov, &bytes, FALSE);
        if (!got && GetLastError() == ERROR_IO_INCOMPLETE)
            return Qnil; /* genuinely still pending — a real timeout */
        /* fall through to settle+stash below */
    } else if (t.wait != WAIT_OBJECT_0) {
        /* WAIT_FAILED (rare): drain so the op is settled, then report. */
        DWORD n = 0;
        CancelIoEx(w->dir, &w->ov);
        GetOverlappedResult(w->dir, &w->ov, &n, TRUE);
        w->armed = 0;
        rearm = watch_arm(w);
        {
            VALUE out = rb_ary_new_capa(3);
            rb_ary_push(out, Qnil);
            rb_ary_push(out, ULONG2NUM(t.gle ? t.gle : ERROR_OPERATION_ABORTED));
            rb_ary_push(out, ULONG2NUM(rearm));
            return out;
        }
    } else {
        /* Signaled: reap the completion (bWait=FALSE, already signaled). */
        got = GetOverlappedResult(w->dir, &w->ov, &bytes, FALSE);
    }

    /* Settle the reaped op into local code/data. */
    w->armed = 0;
    if (got) {
        code = 0;
        if (bytes > 0) {
            data = rb_str_new((const char *)w->buf, (long)bytes);
            rb_enc_associate(data, rb_ascii8bit_encoding());
        }
    } else {
        code = GetLastError(); /* e.g. 995 (aborted by our cancel), 5, 1022, ... */
    }

    /* Re-arm BEFORE returning to Ruby for parsing (rdcw §14), minimizing the
     * no-op-outstanding window. */
    rearm = watch_arm(w);

    /* Interrupt-stash discipline (§5.12): copy the settled batch into the
     * struct, then deliver any pending interrupt. If Thread#kill / Timeout is
     * pending, rb_thread_check_ints() raises here — the batch is safely stashed
     * and the NEXT take returns it (lossless). If nothing is pending we pull the
     * stash back out and return it normally. The raise crosses no live kernel
     * ownership: the op is already re-armed (or recorded un-armed). */
    if (data != Qnil || code != 0) {
        long dlen = (data == Qnil) ? 0 : RSTRING_LEN(data);
        /* No prior stash here (we returned early at the top if one existed). */
        if (dlen > 0) {
            w->stash = (unsigned char *)malloc((size_t)dlen);
            if (w->stash) {
                memcpy(w->stash, RSTRING_PTR(data), (size_t)dlen);
                w->stash_len = (DWORD)dlen;
            } else {
                w->stash_len = 0;
            }
        } else {
            w->stash = NULL;
            w->stash_len = 0;
        }
        w->stash_code = code;

        rb_thread_check_ints(); /* may raise (Thread#kill / Timeout) — batch stashed */

        /* No interrupt: hand the stash back to this caller. */
        {
            VALUE out;
            VALUE sdata = Qnil;
            if (w->stash_len > 0) {
                sdata = rb_str_new((const char *)w->stash, (long)w->stash_len);
                rb_enc_associate(sdata, rb_ascii8bit_encoding());
            }
            code = w->stash_code;
            if (w->stash) { free(w->stash); w->stash = NULL; }
            w->stash_len = 0;
            w->stash_code = 0;
            out = rb_ary_new_capa(3);
            rb_ary_push(out, sdata);
            rb_ary_push(out, ULONG2NUM(code));
            rb_ary_push(out, ULONG2NUM(rearm));
            return out;
        }
    }

    /* data==nil AND code==0: a zero-byte successful completion = kernel buffer
     * overflow (the success spelling, §5.1). The op is re-armed; deliver any
     * pending interrupt, then return [nil, 0, rearm] so Ruby classifies it as
     * :rescan. (Overflow already means "re-enumerate the tree", so the small
     * chance of an interrupt dropping this one rescan signal is harmless — the
     * re-armed op re-signals if still overflowing.) */
    rb_thread_check_ints();
    {
        VALUE out = rb_ary_new_capa(3);
        rb_ary_push(out, Qnil);
        rb_ary_push(out, ULONG2NUM(code));
        rb_ary_push(out, ULONG2NUM(rearm));
        return out;
    }
}

#close ⇒ `Object`

Cancel the in-flight op, close the directory handle, and wake any blocked take (which raises Closed). Pending not-yet-taken events are discarded. Idempotent; safe from any thread/fiber. After :gone the watcher is already closed; calling close again is a no-op.

# File 'lib/winwatch.rb', line 331

def close
  @mode == :winloop ? close_winloop : close_standalone
  nil
end

#closed? ⇒ `Boolean`

Watcher#closed? — raw getter; works on closed objects.

Returns:

(Boolean)

# File 'ext/winwatch/winwatch.c', line 670

static VALUE
watch_closed_p(VALUE self)
{
    watch_t *w = watch_get(self);
    return (w->closed || w->dir == INVALID_HANDLE_VALUE) ? Qtrue : Qfalse;
}

#each ⇒ `Object`

Yield events one at a time, forever. Returns nil after a :gone event (the terminal event), or when the watcher is closed from elsewhere (it swallows the Closed raised by its own internal take). Block required.

Raises:

(ArgumentError)

# File 'lib/winwatch.rb', line 223

def each
  raise ArgumentError, "Winwatch::Watcher#each requires a block" unless block_given?

  loop do
    batch =
      begin
        take
      rescue Closed
        return nil
      end
    next if batch.nil?

    batch.each do |event|
      yield event
      return nil if event.type == :gone
    end
  end
end

#filter ⇒ `Object`

207	# File 'lib/winwatch.rb', line 207 def filter = @filter

#recursive? ⇒ `Boolean`

Returns:

(Boolean)

206	# File 'lib/winwatch.rb', line 206 def recursive? = @recursive

#take(timeout: nil) ⇒ `Object`

Returns a non-empty Array of Winwatch::Event as soon as any are available, or nil on timeout. timeout: nil = forever; non-negative seconds otherwise (ArgumentError if negative); 0 is a non-blocking poll. Raises Winwatch::Closed if the watcher is (or becomes) closed.

# File 'lib/winwatch.rb', line 215

def take(timeout: nil)
  ms = Winwatch.ms_for(timeout)
  @mode == :winloop ? take_winloop(ms) : take_standalone(ms)
end

Class: Winwatch::Watcher

Overview

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#buffer_size ⇒ Object (readonly)

#mode ⇒ Object (readonly)

#path ⇒ Object (readonly)

Class Method Details

._open(vpath, vfilter, vrec, vbuf, vext) ⇒ Object

Instance Method Details

#_arm ⇒ Object

#_cancel ⇒ Object

#_close_handle ⇒ Object

#_close_standalone ⇒ Object

#_external? ⇒ Boolean

#_handle_value ⇒ Object

#_in_take ⇒ Object

#_issue(vov, vbuf, vlen) ⇒ Object

#_mark_closed ⇒ Object

#_take(vms) ⇒ Object

#close ⇒ Object

#closed? ⇒ Boolean

#each ⇒ Object

#filter ⇒ Object

#recursive? ⇒ Boolean

#take(timeout: nil) ⇒ Object

#buffer_size ⇒ `Object` (readonly)

#mode ⇒ `Object` (readonly)

#path ⇒ `Object` (readonly)

._open(vpath, vfilter, vrec, vbuf, vext) ⇒ `Object`

#_arm ⇒ `Object`

#_cancel ⇒ `Object`

#_close_handle ⇒ `Object`

#_close_standalone ⇒ `Object`

#_external? ⇒ `Boolean`

#_handle_value ⇒ `Object`

#_in_take ⇒ `Object`

#_issue(vov, vbuf, vlen) ⇒ `Object`

#_mark_closed ⇒ `Object`

#_take(vms) ⇒ `Object`

#close ⇒ `Object`

#closed? ⇒ `Boolean`

#each ⇒ `Object`

#filter ⇒ `Object`

#recursive? ⇒ `Boolean`

#take(timeout: nil) ⇒ `Object`