From mboxrd@z Thu Jan 1 00:00:00 1970 Received: from [140.186.70.92] (port=44346 helo=eggs.gnu.org) by lists.gnu.org with esmtp (Exim 4.43) id 1POZz9-0001w9-Fe for qemu-devel@nongnu.org; Fri, 03 Dec 2010 13:03:58 -0500 Received: from Debian-exim by eggs.gnu.org with spam-scanned (Exim 4.71) (envelope-from ) id 1POZz7-0003BO-D0 for qemu-devel@nongnu.org; Fri, 03 Dec 2010 13:03:55 -0500 Received: from e2.ny.us.ibm.com ([32.97.182.142]:37875) by eggs.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1POZz7-0003BE-9S for qemu-devel@nongnu.org; Fri, 03 Dec 2010 13:03:53 -0500 Received: from d01dlp02.pok.ibm.com (d01dlp02.pok.ibm.com [9.56.224.85]) by e2.ny.us.ibm.com (8.14.4/8.13.1) with ESMTP id oB3HlNkG009391 for ; Fri, 3 Dec 2010 12:47:23 -0500 Received: from d01relay05.pok.ibm.com (d01relay05.pok.ibm.com [9.56.227.237]) by d01dlp02.pok.ibm.com (Postfix) with ESMTP id 08E2E4DE803B for ; Fri, 3 Dec 2010 13:02:10 -0500 (EST) Received: from d01av01.pok.ibm.com (d01av01.pok.ibm.com [9.56.224.215]) by d01relay05.pok.ibm.com (8.13.8/8.13.8/NCO v10.0) with ESMTP id oB3I3q3J160824 for ; Fri, 3 Dec 2010 13:03:52 -0500 Received: from d01av01.pok.ibm.com (loopback [127.0.0.1]) by d01av01.pok.ibm.com (8.14.4/8.13.1/NCO v10.0 AVout) with ESMTP id oB3I3qM6013217 for ; Fri, 3 Dec 2010 13:03:52 -0500 From: Michael Roth Date: Fri, 3 Dec 2010 12:03:05 -0600 Message-Id: <1291399402-20366-5-git-send-email-mdroth@linux.vnet.ibm.com> In-Reply-To: <1291399402-20366-1-git-send-email-mdroth@linux.vnet.ibm.com> References: <1291399402-20366-1-git-send-email-mdroth@linux.vnet.ibm.com> Subject: [Qemu-devel] [RFC][PATCH v5 04/21] virtagent: transport definitions and job callbacks List-Id: qemu-devel.nongnu.org List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: qemu-devel@nongnu.org Cc: agl@linux.vnet.ibm.com, stefanha@linux.vnet.ibm.com, Jes.Sorensen@redhat.com, mdroth@linux.vnet.ibm.com, aliguori@linux.vnet.ibm.com, ryanh@us.ibm.com, abeekhof@redhat.com Async read/send handlers for managing RPC/HTTP request/responses. The read handler runs continuously and calls into client or server RPC callbacks based on the HTTP header. The send handlers are started up to send RPC client requests/server responses. Signed-off-by: Michael Roth --- virtagent-common.c | 332 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 files changed, 332 insertions(+), 0 deletions(-) diff --git a/virtagent-common.c b/virtagent-common.c index 45f9d9f..1ed2b55 100644 --- a/virtagent-common.c +++ b/virtagent-common.c @@ -89,6 +89,12 @@ typedef struct VAState { static VAState *va_state; +static bool va_set_client_state(enum va_client_state client_state); +static VAServerJob *va_pop_server_job(void); +static VAClientJob *va_pop_client_job(void); +static int va_server_job_add(xmlrpc_mem_block *resp_xml); +static int va_kick(void); + static VAClientJob *va_current_client_job(void) { TRACE("called"); @@ -96,6 +102,332 @@ static VAClientJob *va_current_client_job(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); + + 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) +{ + VAClientJob *client_job; + + client_job = va_pop_client_job(); + assert(client_job != NULL); +TRACE("marker"); + 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) +{ + xmlrpc_mem_block *resp_xml; + VAServerData *server_data = &va_state->server_data; + int ret; + + TRACE("called"); + resp_xml = xmlrpc_registry_process_call(&server_data->env, + server_data->registry, + NULL, content, content_len); + if (resp_xml == NULL) { + LOG("error processing RPC request"); + goto out_bad; + } + + ret = va_server_job_add(resp_xml); + if (ret != 0) { + LOG("error adding server job: %s", strerror(ret)); + } + + return; +out_bad: + /* TODO: should reset state here */ + return; +} + +static void va_http_read_cb(enum va_http_status http_status, + const char *content, size_t content_len, bool is_request) +{ + TRACE("called"); + if (http_status != VA_HTTP_STATUS_OK) { + LOG("error reading http %s", is_request ? "request" : "response"); + content = NULL; + } + + if (is_request) { + va_server_read_cb(content, content_len); + } else { + va_client_read_cb(content, content_len); + } + + return; +} + +/***********************************************************/ +/* utility functions for handling http calls */ + +#define VA_HTTP_REQUEST 1 +#define VA_HTTP_RESPONSE 2 + +static void va_http_hdr_init(VAHTState *s, int request_type) { + const char *preamble; + + TRACE("called"); + /* essentially ignored in the context of virtagent, but might as well */ + if (request_type == VA_HTTP_REQUEST) { + preamble = "POST /RPC2 HTTP/1.1"; + } else if (request_type == VA_HTTP_RESPONSE) { + preamble = "HTTP/1.1 200 OK"; + } else { + s->hdr_len = 0; + return; + } + s->hdr_len = sprintf(s->hdr, + "%s" EOL + "Content-Type: text/xml" EOL + "Content-Length: %u" EOL EOL, + preamble, + (uint32_t)s->content_len); +} + +static void va_rpc_parse_hdr(VAHTState *s) +{ + int i, line_pos = 0; + bool first_line = true; + char line_buf[4096]; + + TRACE("called"); + + for (i = 0; i < VA_HDR_LEN_MAX; ++i) { + if (s->hdr[i] != '\n') { + /* read line */ + line_buf[line_pos++] = s->hdr[i]; + } else { + /* process line */ + if (first_line) { + s->is_request = (strncmp(line_buf, "POST", 4) == 0) ? + true : false; + first_line = false; + } + if (strncmp(line_buf, "Content-Length: ", 16) == 0) { + s->content_len = atoi(&line_buf[16]); + return; + } + line_pos = 0; + } + } +} + +static int va_end_of_header(char *buf, int end_pos) +{ + return !strncmp(buf+(end_pos-2), "\n\r\n", 3); +} + +/***********************************************************/ +/* 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; + + TRACE("called with opaque: %p", opaque); + + switch (s->state) { + case VA_READ_START: + 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; + } 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; + } + 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; + } + s->content_pos += ret; + } + + http_status = VA_HTTP_STATUS_OK; + goto out; + default: + LOG("unknown state"); + goto out_bad; + } + +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->is_request); + /* restart read handler */ + s->state = VA_READ_START; + s->hdr_pos = 0; + s->content_len = 0; + s->content_pos = 0; + qemu_free(s->content); + 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, fd: %d", fd); + + 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; + } + 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; + 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) -- 1.7.0.4