Signed-off-by: Michael Roth <mdr...@linux.vnet.ibm.com>
---
virtagent-common.c | 415 ++++++++++++++++++++++++++++++++++++++++++++++++++++
virtagent-common.h | 1 +
2 files changed, 416 insertions(+), 0 deletions(-)
diff --git a/virtagent-common.c b/virtagent-common.c
index c487252..f8b7d74 100644
--- a/virtagent-common.c
+++ b/virtagent-common.c
@@ -177,6 +177,421 @@ static void va_unset_server_timeout(void)
}
/***********************************************************/
+/* callbacks for read/send handlers */
+
+static void va_client_send_cb(enum va_http_status http_status,
+ const char *content, size_t content_len)
+{
+ VAClientJob *client_job = va_current_client_job();
+
+ TRACE("called");
+ assert(client_job != NULL);
+
+ if (http_status != VA_HTTP_STATUS_OK) {
+ /* TODO: we should reset everything at this point...guest/host will
+ * be out of whack with each other since there's no way to let the
+ * other know job failed (server or client job) if the send channel
+ * is down. But how do we induce the other side to do the same?
+ */
+ LOG("error sending http request");
+ }
+
+ /* request sent ok. free up request xml, then move to
+ * wait (for response) state
+ */
+ XMLRPC_MEMBLOCK_FREE(char, client_job->req_data);
+ assert(va_set_client_state(VA_CLIENT_WAIT));
+}
+
+static void va_server_send_cb(enum va_http_status http_status,
+ const char *content, size_t content_len)
+{
+ VAServerJob *server_job = va_pop_server_job();
+
+ TRACE("called");
+ assert(server_job != NULL);
+ va_unset_server_timeout();
+
+ if (http_status != VA_HTTP_STATUS_OK) {
+ /* TODO: we should reset everything at this point...guest/host will
+ * be out of whack with each other since there's no way to let the
+ * other know job failed (server or client job) if the send channel
+ * is down
+ */
+ LOG("error sending http response");
+ return;
+ }
+
+ /* response sent ok, cleanup server job and kick off the next one */
+ XMLRPC_MEMBLOCK_FREE(char, server_job->resp_data);
+ qemu_free(server_job);
+ va_kick();
+}
+
+static void va_client_read_cb(const char *content, size_t content_len,
+ const char client_tag[64])
+{
+ VAClientJob *client_job;
+
+ TRACE("called");
+ client_job = va_pop_client_job();
+ assert(client_job != NULL);
+ if (--va_state->client_jobs_in_flight == 0) {
+ va_unset_client_timeout();
+ }
+ if (strncmp(client_job->client_tag, client_tag, 64)) {
+ LOG("http client tag mismatch");
+ } else {
+ TRACE("tag matched: %s", client_tag);
+ }
+
+ client_job->cb(content, content_len, client_job->mon_cb,
+ client_job->mon_data);
+ va_kick();
+}
+
+static void va_server_read_cb(const char *content, size_t content_len,
+ const char client_tag[64])
+{
+ int ret;
+
+ TRACE("called");
+ /* generate response and queue it up for sending */
+ ret = va_do_server_rpc(content, content_len, client_tag);
+ if (ret != 0) {
+ LOG("error creating handling remote rpc request: %s", strerror(ret));
+ }
+
+ return;
+}
+
+static void va_http_read_cb(enum va_http_status http_status,
+ const char *content, size_t content_len,
+ const char client_tag[64],
+ enum va_http_type http_type)
+{
+ TRACE("called");
+ if (http_status != VA_HTTP_STATUS_OK) {
+ LOG("error reading http stream (type %d)", http_type);
+ va_cancel_jobs();
+ return;
+ }
+
+ if (http_type == VA_HTTP_TYPE_REQUEST) {
+ TRACE("read request: %s", content);
+ va_server_read_cb(content, content_len, client_tag);
+ } else if (http_type == VA_HTTP_TYPE_RESPONSE) {
+ TRACE("read response: %s", content);
+ va_client_read_cb(content, content_len, client_tag);
+ } else {
+ LOG("unknown http response/request type");
+ va_cancel_jobs();
+ }
+
+ return;
+}
+
+/***********************************************************/
+/* utility functions for handling http calls */
+
+static void va_http_hdr_init(VAHTState *s, enum va_http_type http_type) {
+ const char *preamble;
+
+ TRACE("called");
+ /* essentially ignored in the context of virtagent, but might as well */
+ if (http_type == VA_HTTP_TYPE_REQUEST) {
+ preamble = "POST /RPC2 HTTP/1.1";
+ } else if (http_type == VA_HTTP_TYPE_RESPONSE) {
+ preamble = "HTTP/1.1 200 OK";
+ } else {
+ s->hdr_len = 0;
+ return;
+ }
+ memset(s->hdr, 0, VA_HDR_LEN_MAX);
+ s->hdr_len = sprintf(s->hdr,
+ "%c%s" EOL
+ "Content-Type: text/xml" EOL
+ "Content-Length: %u" EOL
+ "X-Virtagent-Client-Tag: %s" EOL EOL,
+ VA_SENTINEL,
+ preamble,
+ (uint32_t)s->content_len,
+ s->hdr_client_tag[0] ? s->hdr_client_tag : "none");
+}
+
+#define VA_LINE_LEN_MAX 1024
+static void va_rpc_parse_hdr(VAHTState *s)
+{
+ int i, line_pos = 0;
+ bool first_line = true;
+ char line_buf[VA_LINE_LEN_MAX];
+
+ TRACE("called");
+
+ for (i = 0; i < VA_HDR_LEN_MAX; ++i) {
+ if (s->hdr[i] == 0) {
+ /* end of header */
+ return;
+ }
+ if (s->hdr[i] != '\n') {
+ /* read line */
+ line_buf[line_pos++] = s->hdr[i];
+ } else {
+ /* process line */
+ if (first_line) {
+ if (strncmp(line_buf, "POST", 4) == 0) {
+ s->http_type = VA_HTTP_TYPE_REQUEST;
+ } else if (strncmp(line_buf, "HTTP", 4) == 0) {
+ s->http_type = VA_HTTP_TYPE_RESPONSE;
+ } else {
+ s->http_type = VA_HTTP_TYPE_UNKNOWN;
+ }
+ first_line = false;
+ }
+ if (strncmp(line_buf, "Content-Length: ", 16) == 0) {
+ s->content_len = atoi(&line_buf[16]);
+ }
+ if (strncmp(line_buf, "X-Virtagent-Client-Tag: ", 24) == 0) {
+ memcpy(s->hdr_client_tag, &line_buf[24], MIN(line_pos-25, 64));
+ //pstrcpy(s->hdr_client_tag, 64, &line_buf[24]);
+ TRACE("\nTAG<%s>\n", s->hdr_client_tag);
+ }
+ line_pos = 0;
+ memset(line_buf, 0, VA_LINE_LEN_MAX);
+ }
+ }
+}
+
+static int va_end_of_header(char *buf, int end_pos)
+{
+ return !strncmp(buf+(end_pos-2), "\n\r\n", 3);
+}
+
+static void va_http_read_handler_reset(void)
+{
+ VAHTState *s = &va_state->read_state;
+ TRACE("called");
+ s->state = VA_READ_START;
+ s->http_type = VA_HTTP_TYPE_UNKNOWN;
+ s->hdr_pos = 0;
+ s->content_len = 0;
+ s->content_pos = 0;
+ strcpy(s->hdr_client_tag, "none");
+ if (s->content != NULL) {
+ qemu_free(s->content);
+ }
+ s->content = NULL;
+}
+
+/***********************************************************/
+/* read/send handlers */
+
+static void va_http_read_handler(void *opaque)
+{
+ VAHTState *s = &va_state->read_state;
+ enum va_http_status http_status;
+ int fd = va_state->fd;
+ int ret;
+ uint8_t tmp;
+ static int bytes_skipped = 0;
+
+ TRACE("called with opaque: %p", opaque);
+
+ /* until timeouts are implemented, make sure we kick so any deferred
+ * jobs get a chance to run
+ */
+ va_kick();
+
+ switch (s->state) {
+ case VA_READ_START:
+ /* we may have gotten here due to a http error, indicating
+ * a potential unclean state where we are not 'aligned' on http
+ * boundaries. we should read till we hit the next http preamble
+ * rather than assume we're at the start of an http header. since
+ * we control the transport layer on both sides, we'll use a
+ * more reliable sentinal character to mark/detect the start of
+ * the header
+ */
+ while((ret = read(fd, &tmp, 1) > 0) > 0) {
+ if (tmp == VA_SENTINEL) {
+ break;
+ }
+ bytes_skipped += ret;
+ }
+ if (ret == -1) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
+ return;
+ } else {
+ LOG("error reading connection: %s", strerror(errno));
+ goto out_bad;
+ }
+ } else if (ret == 0) {
+ LOG("connected closed unexpectedly");
+ goto out_bad_wait;
+ } else {
+ TRACE("found header, number of bytes skipped: %d",
+ bytes_skipped);
+ bytes_skipped = 0;
+ s->state = VA_READ_HDR;
+ }
+ case VA_READ_HDR:
+ while((ret = read(fd, s->hdr + s->hdr_pos, 1)) > 0
+ && s->hdr_pos < VA_HDR_LEN_MAX) {
+ s->hdr_pos += ret;
+ if (va_end_of_header(s->hdr, s->hdr_pos - 1)) {
+ break;
+ }
+ }
+ if (ret == -1) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
+ return;
+ } else {
+ LOG("error reading connection: %s", strerror(errno));
+ goto out_bad;
+ }
+ } else if (ret == 0) {
+ LOG("connected closed unexpectedly");
+ goto out_bad_wait;
+ } else if (s->hdr_pos >= VA_HDR_LEN_MAX) {
+ LOG("http header too long");
+ goto out_bad;
+ } else {
+ s->content_len = -1;
+ va_rpc_parse_hdr(s);
+ if (s->content_len == -1) {
+ LOG("malformed http header");
+ goto out_bad;
+ } else if (s->content_len > VA_CONTENT_LEN_MAX) {
+ LOG("http content length too long");
+ goto out_bad;
+ }
+ s->content = qemu_mallocz(s->content_len);
+ s->state = VA_READ_BODY;
+ TRACE("read http header:\n<<<%s>>>\n", s->hdr);
+ }
+ case VA_READ_BODY:
+ while(s->content_pos < s->content_len) {
+ ret = read(fd, s->content + s->content_pos,
+ s->content_len - s->content_pos);
+ if (ret == -1) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK
+ || errno == EINTR) {
+ return;
+ } else {
+ LOG("error reading connection: %s", strerror(errno));
+ goto out_bad;
+ }
+ } else if (ret == 0) {
+ LOG("connection closed unexpectedly:"
+ " read %u bytes, expected %u bytes",
+ (unsigned int)s->content_pos, (unsigned
int)s->content_len);
+ goto out_bad_wait;
+ }
+ s->content_pos += ret;
+ }
+
+ TRACE("read http content:\n<<<%s>>>\n", s->content);
+ http_status = VA_HTTP_STATUS_OK;
+ goto out;
+ default:
+ LOG("unknown state");
+ goto out_bad;
+ }
+
+out_bad_wait:
+ /* We should only ever get a read = 0 if we're using virtio and the host
+ * is not connected. this would cause a guest to spin, and we can't do
+ * any work in the meantime, so sleep for a bit here. We also know we
+ * may go ahead and cancel any outstanding jobs at this point, though it
+ * should be noted that we're still ultimately reliant on per-job timeouts
+ * since we might not read EOF before host reconnect.
+ */
+ if (!va_state->is_host &&
+ strcmp(va_state->channel_method, "virtio-serial") == 0) {
+ usleep(100 * 1000);
+ }
+out_bad:
+ http_status = VA_HTTP_STATUS_ERROR;
+out:
+ /* handle the response or request we just read */
+ s->read_cb(http_status, s->content, s->content_len, s->hdr_client_tag,
+ s->http_type);
+ /* restart read handler */
+ va_http_read_handler_reset();
+ http_status = VA_HTTP_STATUS_NEW;
+}
+
+static void va_http_send_handler(void *opaque)
+{
+ VAHTState *s = &va_state->send_state;
+ enum va_http_status http_status;
+ int fd = va_state->fd;
+ int ret;
+
+ TRACE("called");
+
+ switch (s->state) {
+ case VA_SEND_START:
+ s->state = VA_SEND_HDR;
+ case VA_SEND_HDR:
+ do {
+ ret = write(fd, s->hdr + s->hdr_pos, s->hdr_len - s->hdr_pos);
+ if (ret <= 0) {
+ break;
+ }
+ s->hdr_pos += ret;
+ } while (s->hdr_pos < s->hdr_len);
+ if (ret == -1) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
+ return;
+ } else {
+ LOG("error writing header: %s", strerror(errno));
+ goto out_bad;
+ }
+ } else if (ret == 0) {
+ LOG("connected closed unexpectedly");
+ goto out_bad;
+ } else {
+ s->state = VA_SEND_BODY;
+ TRACE("sent http header:\n<<<%s>>>", s->hdr);
+ }
+ case VA_SEND_BODY:
+ do {
+ ret = write(fd, s->content + s->content_pos,
+ s->content_len - s->content_pos);
+ if (ret <= 0) {
+ break;
+ }
+ s->content_pos += ret;
+ } while (s->content_pos < s->content_len);
+ if (ret == -1) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
+ return;
+ } else {
+ LOG("error writing content: %s", strerror(errno));
+ goto out_bad;
+ }
+ } else if (ret == 0) {
+ LOG("connected closed unexpectedly");
+ goto out_bad;
+ } else {
+ http_status = VA_HTTP_STATUS_OK;
+ TRACE("set http content:\n<<<%s>>>", s->content);
+ goto out;
+ }
+ default:
+ LOG("unknown state");
+ goto out_bad;
+ }
+
+out_bad:
+ http_status = VA_HTTP_STATUS_ERROR;
+out:
+ s->send_cb(http_status, s->content, s->content_len);
+ qemu_set_fd_handler(fd, va_http_read_handler, NULL, NULL);
+}
+
+/***********************************************************/
/* functions for starting/managing client/server rpc jobs */
static int va_send_server_response(VAServerJob *server_job)
diff --git a/virtagent-common.h b/virtagent-common.h
index 568df5a..6ad8036 100644
--- a/virtagent-common.h
+++ b/virtagent-common.h
@@ -50,6 +50,7 @@
#define VA_SERVER_JOBS_MAX 5 /* max server rpcs we can queue */
#define VA_SERVER_TIMEOUT_MS 5 * 1000
#define VA_CLIENT_TIMEOUT_MS 5 * 1000
+#define VA_SENTINEL 0xFF
typedef struct VAContext {
bool is_host;
--
1.7.0.4
