X Test that sysread(/head/watch) can be interrupted

wcfs/testprog/wcfs_readcancel.py

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright (C) 2019  Nexedi SA and Contributors.
+#                     Kirill Smelkov <kirr@nexedi.com>
+#
+# This program is free software: you can Use, Study, Modify and Redistribute
+# it under the terms of the GNU General Public License version 3, or (at your
+# option) any later version, as published by the Free Software Foundation.
+#
+# You can also Link and Combine this program with other software covered by
+# the terms of any of the Free Software licenses or any of the Open Source
+# Initiative approved licenses and Convey the resulting work. Corresponding
+# source of such a combination shall include the source code for all other
+# software used.
+#
+# This program is distributed WITHOUT ANY WARRANTY; without even the implied
+# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+#
+# See COPYING file for full licensing terms.
+# See https://www.nexedi.com/licensing for rationale and options.
+"""Program wcfs_readcancel is helper for wcfs_test to verify that
+sysread(/head/watch) is unblocked and canceled when kernel asks WCFS to cancel
+that read request.
+
+Without proper FUSE INTERRUPT handling on WCFS side, such reads are not
+cancelled, which results in processes that were abort or even kill-9`ed being
+stuck forever waiting for WCFS to release them.
+"""
+
+from __future__ import print_function, absolute_import
+
+from golang import select, default
+from golang import context, sync, time
+
+import os, sys
+
+def main():
+    wcfs_root = sys.argv[1]
+    #print("\n\n\n   --------\n\n\n")
+    f = open("%s/head/watch" % wcfs_root)
+    wg = sync.WorkGroup(context.background())
+    def _(ctx):
+        data = f.read()    # should block forever
+        raise AssertionError("read: woken up: data=%r" % data)
+    wg.go(_)
+    def _(ctx):
+        time.sleep(100*time.millisecond)
+        _, _rx = select(
+            default,            # 0
+            ctx.done().recv,    # 1
+        )
+        if _ == 1:
+            raise ctx.err()
+        os._exit(0)
+    wg.go(_)
+    wg.wait()
+
+
+if __name__ == '__main__':
+    main()

wcfs/wcfs_test.py

@@ -39,7 +39,7 @@ from persistent import Persistent
 from persistent.timestamp import TimeStamp
 from ZODB.utils import z64, u64, p64
 
-import sys, os, os.path, subprocess, inspect, traceback
+import sys, os, os.path, subprocess
 from thread import get_ident as gettid
 from time import gmtime
 from errno import EINVAL, ENOENT, ENOTCONN
@@ -50,12 +50,10 @@ from golang import context, sync, time
 from zodbtools.util import ashex as h, fromhex
 import pytest; xfail = pytest.mark.xfail
 from pytest import raises, fail
-#from six import reraise
 from .internal import io, mm
 from .internal._wcfs import _tpywlinkwrite as _twlinkwrite
 from .internal.wcfs_test import _tDB, read_nogil, install_sigbus_trap, fadvise_dontneed
 
-# XXX `py.test -v` -> WENDELIN_CORE_WCFS_OPTIONS += -v=1?
 
 # setup:
 # - create test database, compute zurl and mountpoint for wcfs
@@ -173,8 +171,8 @@ def timeout(parent=context.background()):   # -> ctx
     ctx, _ = with_timeout()
     return ctx
 
-# tdelay is used in places where we need to delay a bit in order to e.g. raise
-# probability of a bug due to race condition.
+# tdelay is used in places where we need to delay a bit in order to e.g.
+# increase probability of a bug due to race condition.
 def tdelay():
     time.sleep(10*time.millisecond)
 
@@ -1072,6 +1070,7 @@ def iter_revv(t, start=z64, level=0):
             yield ([dF.rev] + tail)
 
 
+# -------------------------------------
 # ---- actual tests to access data ----
 
 # exercise wcfs functionality without wcfs invalidation protocol.
@@ -1203,12 +1202,21 @@ def test_wcfs_basic_read_aftertail():
     assert _(100*blksize)   == b''
 
 
+
+# ---- verify wcfs functionality that depends on isolation protocol ----
+
 # verify that watch setup is robust to client errors/misbehaviour.
 @func
 def test_wcfs_watch_robust():
     t = tDB(); zf = t.zfile
     defer(t.close)
 
+    # sysread(/head/watch) can be interrupted
+    p = subprocess.Popen(["%s/testprog/wcfs_readcancel.py" %
+                                os.path.dirname(__file__), t.wc.mountpoint])
+    procwait(timeout(), p)
+
+
     at1 = t.commit(zf, {2:'c1'})
     at2 = t.commit(zf, {2:'c2'})
 
@@ -1353,10 +1361,6 @@ def test_wcfs_pintimeout_kill():
     wg.wait()
 
 
-# TODO verify: a process is blocked in sysread(/head/watch) and is killed -
-# WCFS handles FUSE interrupt request and lets the process go away (if wcfs
-# does not handle interrupt - the process will get stuck)
-
 # watch with @at > head - must wait for head to become >= at.
 # XXX too far ahead - reject?
 @func