You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by as...@apache.org on 2021/03/02 04:24:35 UTC
[qpid-proton] branch master updated (6353ad9 -> 2381e1f)
This is an automated email from the ASF dual-hosted git repository.
astitcher pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git.
from 6353ad9 PROTON-2334: fix libuv proactor PN_PROACTOR_INACTIVE event generation
new e482440 PROTON-2338: [proactor] Add prototypes for new raw connection half close operations
new addf35b PROTON-2338: Implement proactor raw connection half close operations
new 2381e1f PROTON-2338:: Added some testing for raw connection half close
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
c/examples/raw_connect.c | 4 +-
c/examples/raw_echo.c | 6 +-
c/include/proton/raw_connection.h | 36 +++-
c/src/proactor/epoll_raw_connection.c | 33 +++-
c/src/proactor/libuv.c | 2 +
c/src/proactor/raw_connection-internal.h | 36 +++-
c/src/proactor/raw_connection.c | 287 ++++++++++++++++++++++++++-----
c/src/proactor/win_iocp.cpp | 2 +
c/tests/raw_connection_test.cpp | 33 +++-
9 files changed, 383 insertions(+), 56 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-proton] 03/03: PROTON-2338:: Added some testing for raw
connection half close
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
astitcher pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
commit 2381e1f532756a2b3c02fbaa30965fdd14bb1c18
Author: Andrew Stitcher <as...@apache.org>
AuthorDate: Tue Feb 23 17:14:11 2021 -0500
PROTON-2338:: Added some testing for raw connection half close
Unit test half close
Half close write side in raw_connect when writing final message
---
c/examples/raw_connect.c | 4 +++-
c/examples/raw_echo.c | 6 +++++-
c/tests/raw_connection_test.cpp | 33 ++++++++++++++++++++++++++-------
3 files changed, 34 insertions(+), 9 deletions(-)
diff --git a/c/examples/raw_connect.c b/c/examples/raw_connect.c
index 36f2f71..d380801 100644
--- a/c/examples/raw_connect.c
+++ b/c/examples/raw_connect.c
@@ -143,6 +143,7 @@ static void handle_send(app_data_t* app, pn_event_t* event) {
pn_raw_connection_set_context(c, NULL);
free(cd);
printf("**raw connection disconnected\n");
+ pn_proactor_cancel_timeout(app->proactor);
app->disconnects++;
check_condition(event, pn_raw_connection_condition(c), app);
} break;
@@ -153,7 +154,8 @@ static void handle_send(app_data_t* app, pn_event_t* event) {
if (fgets(line, sizeof(line), stdin)) {
send_message(c, line);
} else {
- /* On end of file wait 2 sec for response */
+ /* On end of file, close for write then wait 2 sec for response */
+ pn_raw_connection_write_close(c);
app->towake = c;
pn_proactor_set_timeout(app->proactor, 2000);
}
diff --git a/c/examples/raw_echo.c b/c/examples/raw_echo.c
index 374ffaa..d47f95b 100644
--- a/c/examples/raw_echo.c
+++ b/c/examples/raw_echo.c
@@ -96,7 +96,11 @@ static void send_message(pn_raw_connection_t *c, const char* msg) {
buffer.capacity = 1024;
buffer.offset = 0;
buffer.size = len;
- pn_raw_connection_write_buffers(c, &buffer, 1);
+ // If message not accepted just throw it away!
+ if (pn_raw_connection_write_buffers(c, &buffer, 1) < 1) {
+ printf("**Couldn't send message: write not accepted**\n");
+ free(buf);
+ }
}
static void recv_message(pn_raw_buffer_t buf) {
diff --git a/c/tests/raw_connection_test.cpp b/c/tests/raw_connection_test.cpp
index 4ad4689..b7ce803 100644
--- a/c/tests/raw_connection_test.cpp
+++ b/c/tests/raw_connection_test.cpp
@@ -530,7 +530,7 @@ TEST_CASE("raw connection") {
// No need buffers event as we already gave buffers
REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_EVENT_NONE);
- pni_raw_close(p);
+ pni_raw_write_close(p);
size_t rgiven = pn_raw_connection_take_read_buffers(p, &read[0], rtaken);
REQUIRE(pni_raw_validate(p));
@@ -539,19 +539,35 @@ TEST_CASE("raw connection") {
REQUIRE(pni_raw_validate(p));
CHECK(wgiven==0);
- REQUIRE(pn_raw_connection_is_read_closed(p));
REQUIRE(pn_raw_connection_is_write_closed(p));
- REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_RAW_CONNECTION_CLOSED_READ);
- REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_EVENT_NONE);
-
- REQUIRE_FALSE(pni_raw_can_read(p));
REQUIRE(pni_raw_can_write(p));
pni_raw_write(p, fds[0], snd, set_write_error);
REQUIRE(pni_raw_validate(p));
CHECK(write_err == 0);
+ REQUIRE_FALSE(pni_raw_can_write(p));
+
+ REQUIRE(pni_raw_can_read(p));
+ pni_raw_read(p, fds[0], rcv, set_read_error);
+ REQUIRE(pni_raw_validate(p));
+ CHECK(read_err == 0);
+
REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_RAW_CONNECTION_WRITTEN);
+
+ SECTION("Read close after write close") {
+ pni_raw_read_close(p);
+ }
+
+ SECTION("Full close after write close") {
+ // We should be able to fully close here (even if we read close would be more specific)
+ pni_raw_close(p);
+ }
+
+ REQUIRE(pn_raw_connection_is_read_closed(p));
+ REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_RAW_CONNECTION_CLOSED_READ);
+
+ REQUIRE_FALSE(pni_raw_can_read(p));
REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_RAW_CONNECTION_CLOSED_WRITE);
REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_RAW_CONNECTION_READ);
rgiven = pn_raw_connection_take_read_buffers(p, &read[0], rtaken);
@@ -561,6 +577,10 @@ TEST_CASE("raw connection") {
wgiven = pn_raw_connection_take_written_buffers(p, &written[0], wtaken);
REQUIRE(pni_raw_validate(p));
CHECK(wgiven==wtaken);
+
+ // This should have no affect because we are already read and write closed
+ pni_raw_close(p);
+
REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_RAW_CONNECTION_DISCONNECTED);
REQUIRE(pn_event_type(pni_raw_event_next(p)) == PN_EVENT_NONE);
@@ -770,7 +790,6 @@ TEST_CASE("raw connection") {
REQUIRE(pni_raw_validate(p));
CHECK(rgiven > 0);
- CHECK(pn_raw_connection_read_buffers_capacity(p) == rgiven);
CHECK(read[rgiven-1].size == 0);
// At this point we should have read everything - make sure it matches
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-proton] 01/03: PROTON-2338: [proactor] Add prototypes for new
raw connection half close operations
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
astitcher pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
commit e482440730a47e737214b47ff46a5d309f28324a
Author: Andrew Stitcher <as...@apache.org>
AuthorDate: Mon Feb 1 20:39:06 2021 -0500
PROTON-2338: [proactor] Add prototypes for new raw connection half close operations
- Add stubs for libuv and win_iocp proactor implementations
---
c/include/proton/raw_connection.h | 36 +++++++++++++++++++++++++++++++++++-
c/src/proactor/libuv.c | 2 ++
c/src/proactor/win_iocp.cpp | 2 ++
3 files changed, 39 insertions(+), 1 deletion(-)
diff --git a/c/include/proton/raw_connection.h b/c/include/proton/raw_connection.h
index 7be7e61..d19ff6d 100644
--- a/c/include/proton/raw_connection.h
+++ b/c/include/proton/raw_connection.h
@@ -91,7 +91,9 @@ PNP_EXTERN const struct pn_netaddr_t *pn_raw_connection_remote_addr(pn_raw_conne
/**
* Close a raw connection.
- * This will close the underlying socket and release all buffers held by the raw connection.
+ * This will flush any buffers to be written; close the underlying socket and release all buffers held by the
+ * raw connection.
+ *
* It will cause @ref PN_RAW_CONNECTION_READ and @ref PN_RAW_CONNECTION_WRITTEN to be emitted so
* the application can clean up buffers given to the raw connection. After that a
* @ref PN_RAW_CONNECTION_DISCONNECTED event will be emitted to allow the application to clean up
@@ -101,6 +103,38 @@ PNP_EXTERN const struct pn_netaddr_t *pn_raw_connection_remote_addr(pn_raw_conne
PNP_EXTERN void pn_raw_connection_close(pn_raw_connection_t *connection);
/**
+ * Shutdown a raw connection for reading.
+ * This will close the underlying socket for reading and release all empty read buffers held by the raw connection.
+ *
+ * It will cause @ref PN_RAW_CONNECTION_READ to be emitted so the application can clean up buffers given to the raw
+ * connection. Note that these buffers may still also contain data read from the socket but not yet consumed by the
+ * application.
+ *
+ * If @ref pn_raw_connection_write_close() has already been called then @ref PN_RAW_CONNECTION_DISCONNECTED will then
+ * also be emitted.
+ *
+ * In order to fully close a raw connection the application will need to either call @ref pn_raw_connection_close()
+ * or @ref pn_raw_connection_write_close() after it calls @ref pn_raw_connection_read_close().
+ */
+PNP_EXTERN void pn_raw_connection_read_close(pn_raw_connection_t *connection);
+
+/**
+ * Shutdown a raw connection for writing.
+ * This will flush any buffers to be written to the socket; close the underlying socket for writing and release all
+ * write buffers held by the raw connection.
+ *
+ * It will cause @ref PN_RAW_CONNECTION_WRITTEN to be emitted so the application can clean up write buffers given to
+ * the raw connection.
+ *
+ * If @ref pn_raw_connection_read_close() has already been called then @ref PN_RAW_CONNECTION_DISCONNECTED will then
+ * also be emitted.
+ *
+ * In order to fully close a raw connection the application will need to either call @ref pn_raw_connection_close()
+ * or @ref pn_raw_connection_read_close() after it calls @ref pn_raw_connection_write_close().
+ */
+PNP_EXTERN void pn_raw_connection_write_close(pn_raw_connection_t *connection);
+
+/**
* Query the raw connection for how many more read buffers it can be given
*/
PNP_EXTERN size_t pn_raw_connection_read_buffers_capacity(pn_raw_connection_t *connection);
diff --git a/c/src/proactor/libuv.c b/c/src/proactor/libuv.c
index b06b1eb..3e50685 100644
--- a/c/src/proactor/libuv.c
+++ b/c/src/proactor/libuv.c
@@ -1367,5 +1367,7 @@ void pn_proactor_raw_connect(pn_proactor_t *p, pn_raw_connection_t *rc, const ch
void pn_listener_raw_accept(pn_listener_t *l, pn_raw_connection_t *rc) {}
void pn_raw_connection_wake(pn_raw_connection_t *conn) {}
void pn_raw_connection_close(pn_raw_connection_t *conn) {}
+void pn_raw_connection_read_close(pn_raw_connection_t *conn) {}
+void pn_raw_connection_write_close(pn_raw_connection_t *conn) {}
const struct pn_netaddr_t *pn_raw_connection_local_addr(pn_raw_connection_t *connection) { return NULL; }
const struct pn_netaddr_t *pn_raw_connection_remote_addr(pn_raw_connection_t *connection) { return NULL; }
diff --git a/c/src/proactor/win_iocp.cpp b/c/src/proactor/win_iocp.cpp
index a71947a..51512bd 100644
--- a/c/src/proactor/win_iocp.cpp
+++ b/c/src/proactor/win_iocp.cpp
@@ -3427,5 +3427,7 @@ void pn_proactor_raw_connect(pn_proactor_t *p, pn_raw_connection_t *rc, const ch
void pn_listener_raw_accept(pn_listener_t *l, pn_raw_connection_t *rc) {}
void pn_raw_connection_wake(pn_raw_connection_t *conn) {}
void pn_raw_connection_close(pn_raw_connection_t *conn) {}
+void pn_raw_connection_read_close(pn_raw_connection_t *conn) {}
+void pn_raw_connection_write_close(pn_raw_connection_t *conn) {}
const struct pn_netaddr_t *pn_raw_connection_local_addr(pn_raw_connection_t *connection) { return NULL; }
const struct pn_netaddr_t *pn_raw_connection_remote_addr(pn_raw_connection_t *connection) { return NULL; }
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-proton] 02/03: PROTON-2338: Implement proactor raw connection
half close operations
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
astitcher pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
commit addf35b355983f00057530447d235d1521426aa9
Author: Andrew Stitcher <as...@apache.org>
AuthorDate: Thu Feb 11 21:14:52 2021 -0500
PROTON-2338: Implement proactor raw connection half close operations
The raw connection state machine has been rejigged to encompass half closed
states. The previous separate read/write state booleans have been
combined into a single state. This should allow for clearer connection
state tracking.
---
c/src/proactor/epoll_raw_connection.c | 33 +++-
c/src/proactor/raw_connection-internal.h | 36 +++-
c/src/proactor/raw_connection.c | 287 ++++++++++++++++++++++++++-----
3 files changed, 310 insertions(+), 46 deletions(-)
diff --git a/c/src/proactor/epoll_raw_connection.c b/c/src/proactor/epoll_raw_connection.c
index 8722ff0..88410e3 100644
--- a/c/src/proactor/epoll_raw_connection.c
+++ b/c/src/proactor/epoll_raw_connection.c
@@ -272,14 +272,33 @@ void pn_raw_connection_wake(pn_raw_connection_t *rc) {
if (notify) notify_poller(prc->task.proactor);
}
-void pn_raw_connection_close(pn_raw_connection_t *rc) {
+static inline void set_closed(pn_raw_connection_t *rc)
+{
praw_connection_t *prc = containerof(rc, praw_connection_t, raw_connection);
lock(&prc->task.mutex);
prc->task.closing = true;
unlock(&prc->task.mutex);
+}
+
+void pn_raw_connection_close(pn_raw_connection_t *rc) {
+ set_closed(rc);
pni_raw_close(rc);
}
+void pn_raw_connection_read_close(pn_raw_connection_t *rc) {
+ if (pn_raw_connection_is_write_closed(rc)) {
+ set_closed(rc);
+ }
+ pni_raw_read_close(rc);
+}
+
+void pn_raw_connection_write_close(pn_raw_connection_t *rc) {
+ if (pn_raw_connection_is_read_closed(rc)) {
+ set_closed(rc);
+ }
+ pni_raw_write_close(rc);
+}
+
static pn_event_t *pni_raw_batch_next(pn_event_batch_t *batch) {
praw_connection_t *rc = containerof(batch, praw_connection_t, batch);
pn_raw_connection_t *raw = &rc->raw_connection;
@@ -318,6 +337,14 @@ static long rcv(int fd, void* b, size_t s) {
return recv(fd, b, s, MSG_DONTWAIT);
}
+static int shutr(int fd) {
+ return shutdown(fd, SHUT_RD);
+}
+
+static int shutw(int fd) {
+ return shutdown(fd, SHUT_WR);
+}
+
static void set_error(pn_raw_connection_t *conn, const char *msg, int err) {
psocket_error(containerof(conn, praw_connection_t, raw_connection), err, msg);
}
@@ -372,6 +399,8 @@ void pni_raw_connection_done(praw_connection_t *rc) {
unlock(&rc->task.mutex);
pn_raw_connection_t *raw = &rc->raw_connection;
+ int fd = rc->psocket.epoll_io.fd;
+ pni_raw_process_shutdown(raw, fd, shutr, shutw);
int wanted =
(pni_raw_can_read(raw) ? EPOLLIN : 0) |
(pni_raw_can_write(raw) ? EPOLLOUT : 0);
@@ -379,7 +408,7 @@ void pni_raw_connection_done(praw_connection_t *rc) {
rc->psocket.epoll_io.wanted = wanted;
rearm_polling(&rc->psocket.epoll_io, p->epollfd); // TODO: check for error
} else {
- bool finished_disconnect = raw->rclosed && raw->wclosed && !ready && !raw->disconnectpending;
+ bool finished_disconnect = raw->state==conn_fini && !ready && !raw->disconnectpending;
if (finished_disconnect) {
// If we're closed and we've sent the disconnect then close
pni_raw_finalize(raw);
diff --git a/c/src/proactor/raw_connection-internal.h b/c/src/proactor/raw_connection-internal.h
index 2e183a0..beda92d 100644
--- a/c/src/proactor/raw_connection-internal.h
+++ b/c/src/proactor/raw_connection-internal.h
@@ -39,6 +39,29 @@ typedef enum {
buff_written = 6
} buff_type;
+/*
+ * r = read, w = write
+ * init = initial
+ * o = open - can read/write
+ * d = draining - write draining
+ * c = closing - shutdown pending
+ * s = stopped (closed)
+ */
+typedef enum {
+ conn_wrong = -1,
+ conn_init = 0,
+ conn_ro_wo = 1,
+ conn_ro_wd = 2,
+ conn_ro_wc = 3,
+ conn_ro_ws = 4,
+ conn_rc_wo = 5,
+ conn_rc_wd = 6,
+ conn_rs_wo = 7,
+ conn_rs_wd = 8,
+ conn_rs_ws = 9,
+ conn_fini = 10,
+} raw_conn_state;
+
typedef uint16_t buff_ptr; // This is always the index+1 so that 0 can be special
typedef struct pbuffer_t {
@@ -72,13 +95,13 @@ struct pn_raw_connection_t {
buff_ptr wbuffer_last_towrite;
buff_ptr wbuffer_first_written;
buff_ptr wbuffer_last_written;
- bool rneedbufferevent;
- bool wneedbufferevent;
+
+ uint8_t state; // really raw_conn_state
+ bool rrequestedbuffers;
+ bool wrequestedbuffers;
+
bool rpending;
bool wpending;
- bool rclosed;
- bool wdraining;
- bool wclosed;
bool rclosedpending;
bool wclosedpending;
bool rdrainpending;
@@ -95,8 +118,11 @@ void pni_raw_connected(pn_raw_connection_t *conn);
void pni_raw_connect_failed(pn_raw_connection_t *conn);
void pni_raw_wake(pn_raw_connection_t *conn);
void pni_raw_close(pn_raw_connection_t *conn);
+void pni_raw_read_close(pn_raw_connection_t *conn);
+void pni_raw_write_close(pn_raw_connection_t *conn);
void pni_raw_read(pn_raw_connection_t *conn, int sock, long (*recv)(int, void*, size_t), void (*set_error)(pn_raw_connection_t *, const char *, int));
void pni_raw_write(pn_raw_connection_t *conn, int sock, long (*send)(int, const void*, size_t), void (*set_error)(pn_raw_connection_t *, const char *, int));
+void pni_raw_process_shutdown(pn_raw_connection_t *conn, int sock, int (*shutdown_rd)(int), int (*shutdown_wr)(int));
bool pni_raw_can_read(pn_raw_connection_t *conn);
bool pni_raw_can_write(pn_raw_connection_t *conn);
pn_event_t *pni_raw_event_next(pn_raw_connection_t *conn);
diff --git a/c/src/proactor/raw_connection.c b/c/src/proactor/raw_connection.c
index c88f602..a819b5b 100644
--- a/c/src/proactor/raw_connection.c
+++ b/c/src/proactor/raw_connection.c
@@ -62,6 +62,142 @@ void pni_raw_initialize(pn_raw_connection_t *conn) {
conn->rbuffer_first_empty = 1;
conn->wbuffer_first_empty = 1;
+
+ conn->state = conn_init;
+}
+
+typedef enum {
+ conn_connected = 0,
+ conn_read_closed = 1,
+ conn_write_closed = 2,
+ conn_write_drained = 3,
+ int_read_shutdown = 4,
+ int_write_shutdown = 5,
+ int_disconnect = 6,
+ api_read_close = 7,
+ api_write_close = 8,
+} raw_event;
+
+/*
+ * There's a little trick in this state table - both conn_init - the initial state and the self transition are represented by
+ * 0. This is because no state ever transitions to the initial state.
+ */
+const uint8_t state_transitions[][9] = {
+ /* State\event cconnected, crclosed, cwclosed, cwdrained, irshutdown, iwshutdown, idsconnect, arclose, awclose */
+ [conn_init] = {conn_ro_wo, conn_wrong, conn_wrong, conn_wrong, conn_wrong, conn_wrong, conn_wrong, conn_rc_wo, conn_ro_wd},
+ [conn_ro_wo] = {conn_wrong, conn_rc_wo, conn_ro_wc, 0, conn_wrong, conn_wrong, conn_wrong, conn_rc_wo, conn_ro_wd},
+ [conn_ro_wd] = {conn_wrong, conn_rc_wd, conn_ro_wc, conn_ro_wc, conn_wrong, conn_wrong, conn_wrong, conn_rc_wd, 0},
+ [conn_ro_wc] = {conn_wrong, conn_rs_ws, 0, conn_wrong, conn_wrong, conn_ro_ws, conn_wrong, conn_rs_ws, 0},
+ [conn_ro_ws] = {conn_wrong, conn_rs_ws, 0, 0, conn_wrong, conn_wrong, conn_wrong, conn_rs_ws, 0},
+ [conn_rc_wo] = {conn_wrong, 0, conn_rs_ws, 0, conn_rs_wo, conn_wrong, conn_wrong, 0, conn_rc_wd},
+ [conn_rc_wd] = {conn_wrong, 0, conn_rs_ws, conn_rs_ws, conn_rs_wd, conn_wrong, conn_wrong, 0, 0},
+ [conn_rs_wo] = {conn_wrong, 0, conn_rs_ws, 0, conn_wrong, conn_wrong, conn_wrong, 0, conn_rs_wd},
+ [conn_rs_wd] = {conn_wrong, 0, conn_rs_ws, conn_rs_ws, conn_wrong, conn_wrong, conn_wrong, 0, 0},
+ [conn_rs_ws] = {conn_wrong, 0, 0, conn_wrong, conn_wrong, conn_wrong, conn_fini, 0, 0},
+ [conn_fini] = {conn_wrong, conn_wrong, conn_wrong, conn_wrong, conn_wrong, conn_wrong, 0, 0, 0},
+};
+
+static inline uint8_t pni_raw_new_state(pn_raw_connection_t *conn, raw_event event) {
+ uint8_t old_state = conn->state;
+ uint8_t new_state = state_transitions[old_state][event];
+ assert(new_state != (uint8_t)conn_wrong);
+ return new_state == 0 ? old_state : new_state;
+}
+
+static inline bool pni_raw_ropen(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_ro_wo:
+ case conn_ro_wd:
+ case conn_ro_wc:
+ case conn_ro_ws:
+ return true;
+ default:
+ return false;
+ }
+}
+
+static inline bool pni_raw_rclosed(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_rc_wo:
+ case conn_rc_wd:
+ case conn_rs_wo:
+ case conn_rs_wd:
+ case conn_rs_ws:
+ case conn_fini :
+ return true;
+ default:
+ return false;
+ }
+}
+
+static inline bool pni_raw_rclosing(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_rc_wo:
+ case conn_rc_wd:
+ return true;
+ default:
+ return false;
+ }
+}
+
+static inline bool pni_raw_wopen(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_ro_wo:
+ case conn_rc_wo:
+ case conn_rs_wo:
+ return true;
+ default:
+ return false;
+ }
+}
+
+// Return whether closed for writing by application (name a little confusing!)
+// NB Initial state is NOT closed for writing - app can write before connecting
+static inline bool pni_raw_wclosed(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_ro_wd:
+ case conn_ro_wc:
+ case conn_ro_ws:
+ case conn_rc_wd:
+ case conn_rs_wd:
+ case conn_rs_ws:
+ case conn_fini :
+ return true;
+ default:
+ return false;
+ }
+}
+
+// Return whether closed and all writes drained
+static inline bool pni_raw_wdrained(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_ro_wc:
+ case conn_ro_ws:
+ case conn_rs_ws:
+ case conn_fini :
+ return true;
+ default:
+ return false;
+ }
+}
+
+static inline bool pni_raw_wclosing(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_ro_wc:
+ return true;
+ default:
+ return false;
+ }
+}
+
+static inline bool pni_raw_rwclosed(pn_raw_connection_t *conn) {
+ switch (conn->state) {
+ case conn_rs_ws:
+ case conn_fini :
+ return true;
+ default:
+ return false;
+ }
}
bool pni_raw_validate(pn_raw_connection_t *conn) {
@@ -121,12 +257,14 @@ void pni_raw_finalize(pn_raw_connection_t *conn) {
size_t pn_raw_connection_read_buffers_capacity(pn_raw_connection_t *conn) {
assert(conn);
- return read_buffer_count - conn->rbuffer_count;
+ bool rclosed = pni_raw_rclosed(conn);
+ return rclosed ? 0 : (read_buffer_count - conn->rbuffer_count);
}
size_t pn_raw_connection_write_buffers_capacity(pn_raw_connection_t *conn) {
assert(conn);
- return write_buffer_count-conn->wbuffer_count;
+ bool wclosed = pni_raw_wclosed(conn);
+ return wclosed ? 0 : (write_buffer_count-conn->wbuffer_count);
}
size_t pn_raw_connection_give_read_buffers(pn_raw_connection_t *conn, pn_raw_buffer_t const *buffers, size_t num) {
@@ -160,7 +298,7 @@ size_t pn_raw_connection_give_read_buffers(pn_raw_connection_t *conn, pn_raw_buf
conn->rbuffer_first_empty = current;
conn->rbuffer_count += can_take;
- conn->rneedbufferevent = false;
+ conn->rrequestedbuffers = false;
return can_take;
}
@@ -232,7 +370,7 @@ size_t pn_raw_connection_write_buffers(pn_raw_connection_t *conn, pn_raw_buffer_
conn->wbuffer_first_empty = current;
conn->wbuffer_count += can_take;
- conn->wneedbufferevent = false;
+ conn->wrequestedbuffers = false;
return can_take;
}
@@ -314,16 +452,17 @@ static inline void pni_raw_release_buffers(pn_raw_connection_t *conn) {
static inline void pni_raw_disconnect(pn_raw_connection_t *conn) {
pni_raw_release_buffers(conn);
conn->disconnectpending = true;
+ conn->state = pni_raw_new_state(conn, int_disconnect);
}
void pni_raw_connected(pn_raw_connection_t *conn) {
pn_condition_clear(conn->condition);
pni_raw_put_event(conn, PN_RAW_CONNECTION_CONNECTED);
+ conn->state = pni_raw_new_state(conn, conn_connected);
}
void pni_raw_connect_failed(pn_raw_connection_t *conn) {
- conn->rclosed = true;
- conn->wclosed = true;
+ conn->state = conn_fini;
pni_raw_disconnect(conn);
}
@@ -333,6 +472,9 @@ void pni_raw_wake(pn_raw_connection_t *conn) {
void pni_raw_read(pn_raw_connection_t *conn, int sock, long (*recv)(int, void*, size_t), void(*set_error)(pn_raw_connection_t *, const char *, int)) {
assert(conn);
+
+ if (!pni_raw_ropen(conn)) return;
+
bool closed = false;
for(;conn->rbuffer_first_unused;) {
buff_ptr p = conn->rbuffer_first_unused;
@@ -384,14 +526,18 @@ finished_reading:
if (conn->rbuffer_first_read && !conn->rpending) {
conn->rpending = true;
}
+
+ uint8_t old_state = conn->state;
+
// Socket closed for read
if (closed) {
- if (!conn->rclosed) {
- conn->rclosed = true;
- conn->rclosedpending = true;
- if (conn->wclosed) {
- pni_raw_disconnect(conn);
- }
+ conn->state = pni_raw_new_state(conn, conn_read_closed);
+ conn->rclosedpending = true;
+ }
+
+ if (conn->state != old_state) {
+ if (pni_raw_rwclosed(conn)) {
+ pni_raw_disconnect(conn);
}
}
return;
@@ -399,7 +545,11 @@ finished_reading:
void pni_raw_write(pn_raw_connection_t *conn, int sock, long (*send)(int, const void*, size_t), void(*set_error)(pn_raw_connection_t *, const char *, int)) {
assert(conn);
+
+ if (pni_raw_wdrained(conn)) return;
+
bool closed = false;
+ bool drained = false;
for(;conn->wbuffer_first_towrite;) {
buff_ptr p = conn->wbuffer_first_towrite;
assert(conn->wbuffers[p-1].type == buff_unwritten);
@@ -453,31 +603,52 @@ void pni_raw_write(pn_raw_connection_t *conn, int sock, long (*send)(int, const
finished_writing:
if (!conn->wbuffer_first_towrite) {
conn->wbuffer_last_towrite = 0;
- closed = closed || conn->wdraining;
+ drained = true;
}
// Wrote something; end of stream; out of buffers; or blocked for write
if (conn->wbuffer_first_written && !conn->wpending) {
conn->wpending = true;
}
+
+ uint8_t old_state = conn->state;
+
+ // Drained all writes
+ if (drained) {
+ conn->state = pni_raw_new_state(conn, conn_write_drained);
+ }
+
// Socket closed for write
if (closed) {
- if (!conn->wclosed) {
- conn->wclosed = true;
- conn->wclosedpending = true;
- if (conn->rclosed) {
- pni_raw_disconnect(conn);;
- }
+ conn->state = pni_raw_new_state(conn, conn_write_closed);
+ conn->wclosedpending = true;
+ }
+
+ if (conn->state != old_state) {
+ if (pni_raw_rwclosed(conn)) {
+ pni_raw_disconnect(conn);
}
}
return;
}
+void pni_raw_process_shutdown(pn_raw_connection_t *conn, int sock, int (*shutdown_rd)(int), int (*shutdown_wr)(int)) {
+ assert(conn);
+ if (pni_raw_rclosing(conn)) {
+ shutdown_rd(sock);
+ conn->state = pni_raw_new_state(conn, int_read_shutdown);
+ }
+ if (pni_raw_wclosing(conn)) {
+ shutdown_wr(sock);
+ conn->state = pni_raw_new_state(conn, int_write_shutdown);
+ }
+}
+
bool pni_raw_can_read(pn_raw_connection_t *conn) {
- return !conn->rclosed && conn->rbuffer_first_unused;
+ return pni_raw_ropen(conn) && conn->rbuffer_first_unused;
}
bool pni_raw_can_write(pn_raw_connection_t *conn) {
- return !conn->wclosed && conn->wbuffer_first_towrite;
+ return !pni_raw_wdrained(conn) && conn->wbuffer_first_towrite;
}
pn_event_t *pni_raw_event_next(pn_raw_connection_t *conn) {
@@ -510,50 +681,88 @@ pn_event_t *pni_raw_event_next(pn_raw_connection_t *conn) {
} else if (conn->disconnectpending) {
pni_raw_put_event(conn, PN_RAW_CONNECTION_DISCONNECTED);
conn->disconnectpending = false;
- } else if (!conn->wclosed && !conn->wbuffer_first_towrite && !conn->wneedbufferevent) {
+ } else if (!pni_raw_wdrained(conn) && !conn->wbuffer_first_towrite && !conn->wrequestedbuffers) {
// Ran out of write buffers
pni_raw_put_event(conn, PN_RAW_CONNECTION_NEED_WRITE_BUFFERS);
- conn->wneedbufferevent = true;
- } else if (!conn->rclosed && !conn->rbuffer_first_unused && !conn->rneedbufferevent) {
+ conn->wrequestedbuffers = true;
+ } else if (!pni_raw_rclosed(conn) && !conn->rbuffer_first_unused && !conn->rrequestedbuffers) {
// Ran out of read buffers
pni_raw_put_event(conn, PN_RAW_CONNECTION_NEED_READ_BUFFERS);
- conn->rneedbufferevent = true;
+ conn->rrequestedbuffers = true;
} else {
return NULL;
}
} while (true);
}
+void pni_raw_read_close(pn_raw_connection_t *conn) {
+ // If already fully closed nothing to do
+ if (pni_raw_rwclosed(conn)) {
+ return;
+ }
+ if (!pni_raw_rclosed(conn)) {
+ conn->rclosedpending = true;
+ }
+ conn->state = pni_raw_new_state(conn, api_read_close);
+ if (pni_raw_rwclosed(conn)) {
+ pni_raw_disconnect(conn);
+ }
+}
+
+void pni_raw_write_close(pn_raw_connection_t *conn) {
+ // If already fully closed nothing to do
+ if (pni_raw_rwclosed(conn)) {
+ return;
+ }
+
+ if (!pni_raw_wdrained(conn)) {
+ if (!pni_raw_wclosed(conn)) {
+ conn->wclosedpending = true;
+ }
+ conn->state = pni_raw_new_state(conn, api_write_close);
+
+ if (!conn->wbuffer_first_towrite) {
+ conn->state = pni_raw_new_state(conn, conn_write_drained);
+ }
+ }
+ if (pni_raw_rwclosed(conn)) {
+ pni_raw_disconnect(conn);
+ }
+}
+
void pni_raw_close(pn_raw_connection_t *conn) {
- // TODO: Do we need different flags here?
- // TODO: What is the precise semantics for close?
- bool rclosed = conn->rclosed;
- if (!rclosed) {
- conn->rclosed = true;
+ // If already fully closed nothing to do
+ if (pni_raw_rwclosed(conn)) {
+ return;
+ }
+ if (!pni_raw_rclosed(conn)) {
conn->rclosedpending = true;
}
- bool wclosed = conn->wclosed;
- if (!wclosed) {
- if (conn->wbuffer_first_towrite) {
- conn->wdraining = true;
- } else {
- conn->wclosed = true;
+ conn->state = pni_raw_new_state(conn, api_read_close);
+
+ if (!pni_raw_wdrained(conn)) {
+ if (!pni_raw_wclosed(conn)) {
conn->wclosedpending = true;
}
+ conn->state = pni_raw_new_state(conn, api_write_close);
+
+ if (!conn->wbuffer_first_towrite) {
+ conn->state = pni_raw_new_state(conn, conn_write_drained);
+ }
}
- if ((!rclosed || !wclosed) && !conn->wdraining) {
+ if (pni_raw_rwclosed(conn)) {
pni_raw_disconnect(conn);
}
}
bool pn_raw_connection_is_read_closed(pn_raw_connection_t *conn) {
assert(conn);
- return conn->rclosed;
+ return pni_raw_rclosed(conn);
}
bool pn_raw_connection_is_write_closed(pn_raw_connection_t *conn) {
assert(conn);
- return conn->wclosed || conn->wdraining;
+ return pni_raw_wclosed(conn);
}
pn_condition_t *pn_raw_connection_condition(pn_raw_connection_t *conn) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org