Go: introducing SHM_ACK observer.

Each request processed in a separate goroutine.  In case of OOSM state,
during response write, request goroutine blocks on channel which waits
event from main thread about SHM_ACK message from router.

3 files changed, 77 insertions, 11 deletions

diff --git a/go/nxt_cgo_lib.c b/go/nxt_cgo_lib.c
index 5cb31b5a..a4fef9ea 100644
--- a/go/nxt_cgo_lib.c
+++ b/go/nxt_cgo_lib.c
@@ -19,6 +19,7 @@ static ssize_t nxt_cgo_port_send(nxt_unit_ctx_t *, nxt_unit_port_id_t *port_id,
     const void *buf, size_t buf_size, const void *oob, size_t oob_size);
 static ssize_t nxt_cgo_port_recv(nxt_unit_ctx_t *, nxt_unit_port_id_t *port_id,
     void *buf, size_t buf_size, void *oob, size_t oob_size);
+static void nxt_cgo_shm_ack_handler(nxt_unit_ctx_t *ctx);
 
 int
 nxt_cgo_run(uintptr_t handler)
@@ -34,6 +35,7 @@ nxt_cgo_run(uintptr_t handler)
     init.callbacks.remove_port     = nxt_cgo_remove_port;
     init.callbacks.port_send       = nxt_cgo_port_send;
     init.callbacks.port_recv       = nxt_cgo_port_recv;
+    init.callbacks.shm_ack_handler = nxt_cgo_shm_ack_handler;
 
     init.data = (void *) handler;
 
@@ -137,6 +139,13 @@ nxt_cgo_port_recv(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id,
 }
 
 
+static void
+nxt_cgo_shm_ack_handler(nxt_unit_ctx_t *ctx)
+{
+    return nxt_go_shm_ack_handler();
+}
+
+
 int
 nxt_cgo_response_create(uintptr_t req, int status, int fields,
     uint32_t fields_size)
@@ -166,15 +175,8 @@ nxt_cgo_response_send(uintptr_t req)
 ssize_t
 nxt_cgo_response_write(uintptr_t req, uintptr_t start, uint32_t len)
 {
-    int  rc;
-
-    rc = nxt_unit_response_write((nxt_unit_request_info_t *) req,
-                                 (void *) start, len);
-    if (rc != NXT_UNIT_OK) {
-        return -1;
-    }
-
-    return len;
+    return nxt_unit_response_write_nb((nxt_unit_request_info_t *) req,
+                                      (void *) start, len, 0);
 }
 
 
diff --git a/go/observable.go b/go/observable.go
new file mode 100644
index 00000000..9a38802c
--- /dev/null
+++ b/go/observable.go
@@ -0,0 +1,32 @@
+/*
+ * Copyright (C) NGINX, Inc.
+ */
+
+package unit
+
+import (
+	"sync"
+)
+
+type observable struct {
+	sync.Mutex
+	observers []chan int
+}
+
+func (o *observable) attach(c chan int) {
+	o.Lock()
+	defer o.Unlock()
+
+	o.observers = append(o.observers, c)
+}
+
+func (o *observable) notify(e int) {
+	o.Lock()
+	defer o.Unlock()
+
+	for _, v := range o.observers {
+		v <- e
+	}
+
+	o.observers = nil
+}
diff --git a/go/response.go b/go/response.go
index 767d66b7..bfa79656 100644
--- a/go/response.go
+++ b/go/response.go
@@ -19,6 +19,7 @@ type response struct {
 	headerSent bool
 	req        *http.Request
 	c_req      C.uintptr_t
+	ch         chan int
 }
 
 func new_response(c_req C.uintptr_t, req *http.Request) *response {
@@ -40,8 +41,26 @@ func (r *response) Write(p []byte) (n int, err error) {
 		r.WriteHeader(http.StatusOK)
 	}
 
-	res := C.nxt_cgo_response_write(r.c_req, buf_ref(p), C.uint32_t(len(p)))
-	return int(res), nil
+	l := len(p)
+	written := int(0)
+	br := buf_ref(p)
+
+	for written < l {
+		res := C.nxt_cgo_response_write(r.c_req, br, C.uint32_t(l - written))
+
+		written += int(res)
+		br += C.uintptr_t(res)
+
+		if (written < l) {
+			if r.ch == nil {
+				r.ch = make(chan int, 2)
+			}
+
+			wait_shm_ack(r.ch)
+		}
+	}
+
+	return written, nil
 }
 
 func (r *response) WriteHeader(code int) {
@@ -85,3 +104,16 @@ func (r *response) Flush() {
 		r.WriteHeader(http.StatusOK)
 	}
 }
+
+var observer_registry_ observable
+
+func wait_shm_ack(c chan int) {
+	observer_registry_.attach(c)
+
+	_ = <-c
+}
+
+//export nxt_go_shm_ack_handler
+func nxt_go_shm_ack_handler() {
+	observer_registry_.notify(1)
+}