libvirt/daemon/stream.c

797 lines
23 KiB
C
Raw Normal View History

Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
/*
* stream.c: APIs for managing client streams
*
maint: use consistent if-else braces in remaining spots I'm about to add a syntax check that enforces our documented HACKING style of always using matching {} on if-else statements. This patch focuses on all remaining problems, where there weren't enough issues to warrant splitting it further. * src/remote/remote_driver.c (doRemoteOpen): Correct use of {}. * src/security/virt-aa-helper.c (vah_add_path, valid_path, main): Likewise. * src/rpc/virnetsocket.c (virNetSocketNewConnectLibSSH2): Likewise. * src/esx/esx_vi_types.c (esxVI_Type_FromString): Likewise. * src/uml/uml_driver.c (umlDomainDetachDevice): Likewise. * src/util/viralloc.c (virShrinkN): Likewise. * src/util/virbuffer.c (virBufferURIEncodeString): Likewise. * src/util/virdbus.c (virDBusCall): Likewise. * src/util/virnetdev.c (virNetDevValidateConfig): Likewise. * src/util/virnetdevvportprofile.c (virNetDevVPortProfileGetNthParent): Likewise. * src/util/virpci.c (virPCIDeviceIterDevices) (virPCIDeviceWaitForCleanup) (virPCIDeviceIsBehindSwitchLackingACS): Likewise. * src/util/virsocketaddr.c (virSocketAddrGetNumNetmaskBits): Likewise. * src/util/viruri.c (virURIParseParams): Likewise. * daemon/stream.c (daemonStreamHandleAbort): Likewise. * tests/testutils.c (virtTestResult): Likewise. * tests/cputest.c (cpuTestBaseline): Likewise. * tools/virsh-domain.c (cmdDomPMSuspend): Likewise. * tools/virsh-host.c (cmdNodeSuspend): Likewise. * src/esx/esx_vi_generator.py (Type.generate_typefromstring): Tweak generated code. Signed-off-by: Eric Blake <eblake@redhat.com>
2014-09-03 19:39:21 +00:00
* Copyright (C) 2009-2014 Red Hat, Inc.
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library. If not, see
* <http://www.gnu.org/licenses/>.
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
*
* Author: Daniel P. Berrange <berrange@redhat.com>
*/
#include <config.h>
#include "stream.h"
#include "remote.h"
2012-12-12 18:06:53 +00:00
#include "viralloc.h"
2012-12-12 17:59:27 +00:00
#include "virlog.h"
#include "virnetserverclient.h"
#include "virerror.h"
#define VIR_FROM_THIS VIR_FROM_STREAMS
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
VIR_LOG_INIT("daemon.stream");
struct daemonClientStream {
daemonClientPrivatePtr priv;
int refs;
virNetServerProgramPtr prog;
virStreamPtr st;
int procedure;
unsigned int serial;
bool recvEOF;
bool closed;
int filterID;
virNetMessagePtr rx;
bool tx;
daemonClientStreamPtr next;
};
static int
daemonStreamHandleWrite(virNetServerClientPtr client,
daemonClientStream *stream);
static int
daemonStreamHandleRead(virNetServerClientPtr client,
daemonClientStream *stream);
static int
daemonStreamHandleFinish(virNetServerClientPtr client,
daemonClientStream *stream,
virNetMessagePtr msg);
static int
daemonStreamHandleAbort(virNetServerClientPtr client,
daemonClientStream *stream,
virNetMessagePtr msg);
static void
daemonStreamUpdateEvents(daemonClientStream *stream)
{
int newEvents = 0;
if (stream->closed)
return;
if (stream->rx)
newEvents |= VIR_STREAM_EVENT_WRITABLE;
if (stream->tx && !stream->recvEOF)
newEvents |= VIR_STREAM_EVENT_READABLE;
virStreamEventUpdateCallback(stream->st, newEvents);
}
/*
* Invoked when an outgoing data packet message has been fully sent.
* This simply re-enables TX of further data.
*
* The idea is to stop the daemon growing without bound due to
* fast stream, but slow client
*/
static void
daemonStreamMessageFinished(virNetMessagePtr msg,
void *opaque)
{
daemonClientStream *stream = opaque;
VIR_DEBUG("stream=%p proc=%d serial=%u",
stream, msg->header.proc, msg->header.serial);
stream->tx = true;
daemonStreamUpdateEvents(stream);
daemonFreeClientStream(NULL, stream);
}
/*
* Callback that gets invoked when a stream becomes writable/readable
*/
static void
daemonStreamEvent(virStreamPtr st, int events, void *opaque)
{
virNetServerClientPtr client = opaque;
daemonClientStream *stream;
daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client);
virMutexLock(&priv->lock);
stream = priv->streams;
while (stream) {
if (stream->st == st)
break;
stream = stream->next;
}
if (!stream) {
VIR_WARN("event for client=%p stream st=%p, but missing stream state", client, st);
virStreamEventRemoveCallback(st);
goto cleanup;
}
VIR_DEBUG("st=%p events=%d EOF=%d closed=%d", st, events, stream->recvEOF, stream->closed);
if (!stream->closed &&
(events & VIR_STREAM_EVENT_WRITABLE)) {
if (daemonStreamHandleWrite(client, stream) < 0) {
daemonRemoveClientStream(client, stream);
virNetServerClientClose(client);
goto cleanup;
}
}
if (!stream->closed && !stream->recvEOF &&
(events & (VIR_STREAM_EVENT_READABLE))) {
events = events & ~(VIR_STREAM_EVENT_READABLE);
if (daemonStreamHandleRead(client, stream) < 0) {
daemonRemoveClientStream(client, stream);
virNetServerClientClose(client);
goto cleanup;
}
/* If we detected EOF during read processing,
* then clear hangup/error conditions, since
* we want the client to see the EOF message
* we just sent them
*/
if (stream->recvEOF)
events = events & ~(VIR_STREAM_EVENT_HANGUP |
VIR_STREAM_EVENT_ERROR);
}
/* If we have a completion/abort message, always process it */
if (stream->rx) {
virNetMessagePtr msg = stream->rx;
switch (msg->header.status) {
case VIR_NET_CONTINUE:
/* nada */
break;
case VIR_NET_OK:
virNetMessageQueueServe(&stream->rx);
if (daemonStreamHandleFinish(client, stream, msg) < 0) {
virNetMessageFree(msg);
daemonRemoveClientStream(client, stream);
virNetServerClientClose(client);
goto cleanup;
}
break;
case VIR_NET_ERROR:
default:
virNetMessageQueueServe(&stream->rx);
if (daemonStreamHandleAbort(client, stream, msg) < 0) {
virNetMessageFree(msg);
daemonRemoveClientStream(client, stream);
virNetServerClientClose(client);
goto cleanup;
}
break;
}
}
/* If we got HANGUP, we need to only send an empty
* packet so the client sees an EOF and cleans up
*/
if (!stream->closed && !stream->recvEOF &&
(events & VIR_STREAM_EVENT_HANGUP)) {
virNetMessagePtr msg;
events &= ~(VIR_STREAM_EVENT_HANGUP);
stream->tx = false;
stream->recvEOF = true;
if (!(msg = virNetMessageNew(false))) {
daemonRemoveClientStream(client, stream);
virNetServerClientClose(client);
goto cleanup;
}
msg->cb = daemonStreamMessageFinished;
msg->opaque = stream;
stream->refs++;
if (virNetServerProgramSendStreamData(remoteProgram,
client,
msg,
stream->procedure,
stream->serial,
"", 0) < 0) {
virNetMessageFree(msg);
daemonRemoveClientStream(client, stream);
virNetServerClientClose(client);
goto cleanup;
}
}
if (!stream->closed &&
(events & (VIR_STREAM_EVENT_ERROR | VIR_STREAM_EVENT_HANGUP))) {
int ret;
virNetMessagePtr msg;
virNetMessageError rerr;
memset(&rerr, 0, sizeof(rerr));
stream->closed = true;
virStreamEventRemoveCallback(stream->st);
virStreamAbort(stream->st);
if (events & VIR_STREAM_EVENT_HANGUP)
virReportError(VIR_ERR_RPC,
"%s", _("stream had unexpected termination"));
else
virReportError(VIR_ERR_RPC,
"%s", _("stream had I/O failure"));
2011-08-31 16:42:58 +00:00
msg = virNetMessageNew(false);
if (!msg) {
ret = -1;
} else {
ret = virNetServerProgramSendStreamError(remoteProgram,
client,
msg,
&rerr,
stream->procedure,
stream->serial);
}
daemonRemoveClientStream(client, stream);
if (ret < 0)
virNetServerClientClose(client);
goto cleanup;
}
if (stream->closed) {
daemonRemoveClientStream(client, stream);
} else {
daemonStreamUpdateEvents(stream);
}
cleanup:
virMutexUnlock(&priv->lock);
}
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
/*
* @client: a locked client object
*
* Invoked by the main loop when filtering incoming messages.
*
* Returns 1 if the message was processed, 0 if skipped,
* -1 on fatal client error
*/
static int
daemonStreamFilter(virNetServerClientPtr client ATTRIBUTE_UNUSED,
virNetMessagePtr msg,
void *opaque)
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
{
daemonClientStream *stream = opaque;
int ret = 0;
virMutexLock(&stream->priv->lock);
if (msg->header.type != VIR_NET_STREAM)
goto cleanup;
if (!virNetServerProgramMatches(stream->prog, msg))
goto cleanup;
if (msg->header.proc != stream->procedure ||
msg->header.serial != stream->serial)
goto cleanup;
VIR_DEBUG("Incoming client=%p, rx=%p, serial=%u, proc=%d, status=%d",
client, stream->rx, msg->header.proc,
msg->header.serial, msg->header.status);
virNetMessageQueuePush(&stream->rx, msg);
daemonStreamUpdateEvents(stream);
ret = 1;
cleanup:
virMutexUnlock(&stream->priv->lock);
return ret;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
}
/*
* @conn: a connection object to associate the stream with
* @header: the method call to associate with the stream
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
*
* Creates a new stream for this conn
*
* Returns a new stream object, or NULL upon OOM
*/
daemonClientStream *
daemonCreateClientStream(virNetServerClientPtr client,
virStreamPtr st,
virNetServerProgramPtr prog,
virNetMessageHeaderPtr header)
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
{
daemonClientStream *stream;
daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
VIR_DEBUG("client=%p, proc=%d, serial=%u, st=%p",
client, header->proc, header->serial, st);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
if (VIR_ALLOC(stream) < 0)
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
return NULL;
stream->refs = 1;
stream->priv = priv;
stream->prog = virObjectRef(prog);
stream->procedure = header->proc;
stream->serial = header->serial;
stream->filterID = -1;
stream->st = st;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
return stream;
}
/*
* @stream: an unused client stream
*
* Frees the memory associated with this inactive client
* stream
*/
int daemonFreeClientStream(virNetServerClientPtr client,
daemonClientStream *stream)
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
{
virNetMessagePtr msg;
int ret = 0;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
if (!stream)
return 0;
stream->refs--;
if (stream->refs)
return 0;
VIR_DEBUG("client=%p, proc=%d, serial=%u",
client, stream->procedure, stream->serial);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
virObjectUnref(stream->prog);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
msg = stream->rx;
while (msg) {
virNetMessagePtr tmp = msg->next;
if (client) {
/* Send a dummy reply to free up 'msg' & unblock client rx */
2011-08-31 16:42:58 +00:00
virNetMessageClear(msg);
msg->header.type = VIR_NET_REPLY;
if (virNetServerClientSendMessage(client, msg) < 0) {
virNetServerClientImmediateClose(client);
virNetMessageFree(msg);
ret = -1;
}
} else {
virNetMessageFree(msg);
}
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
msg = tmp;
}
virObjectUnref(stream->st);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
VIR_FREE(stream);
return ret;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
}
/*
* @client: a locked client to add the stream to
* @stream: a stream to add
*/
int daemonAddClientStream(virNetServerClientPtr client,
daemonClientStream *stream,
bool transmit)
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
{
VIR_DEBUG("client=%p, proc=%d, serial=%u, st=%p, transmit=%d",
client, stream->procedure, stream->serial, stream->st, transmit);
daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
if (stream->filterID != -1) {
VIR_WARN("Filter already added to client %p", client);
return -1;
}
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
if (virStreamEventAddCallback(stream->st, 0,
daemonStreamEvent, client,
virObjectFreeCallback) < 0)
return -1;
virObjectRef(client);
if ((stream->filterID = virNetServerClientAddFilter(client,
daemonStreamFilter,
stream)) < 0) {
virStreamEventRemoveCallback(stream->st);
return -1;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
}
if (transmit)
stream->tx = true;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
virMutexLock(&priv->lock);
stream->next = priv->streams;
priv->streams = stream;
daemonStreamUpdateEvents(stream);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
virMutexUnlock(&priv->lock);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
return 0;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
}
/*
* @client: a locked client object
* @stream: an inactive, closed stream object
*
* Removes a stream from the list of active streams for the client
*
* Returns 0 if the stream was removed, -1 if it doesn't exist
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
*/
int
daemonRemoveClientStream(virNetServerClientPtr client,
daemonClientStream *stream)
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
{
VIR_DEBUG("client=%p, proc=%d, serial=%u, st=%p",
client, stream->procedure, stream->serial, stream->st);
daemonClientPrivatePtr priv = virNetServerClientGetPrivateData(client);
daemonClientStream *curr = priv->streams;
daemonClientStream *prev = NULL;
if (stream->filterID != -1) {
virNetServerClientRemoveFilter(client,
stream->filterID);
stream->filterID = -1;
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
}
if (!stream->closed) {
stream->closed = true;
virStreamEventRemoveCallback(stream->st);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
virStreamAbort(stream->st);
}
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
while (curr) {
if (curr == stream) {
if (prev)
prev->next = curr->next;
else
priv->streams = curr->next;
return daemonFreeClientStream(client, stream);
Helper functions for processing data streams in libvirtd Defines the extensions to the remote protocol for generic data streams. Adds a bunch of helper code to the libvirtd daemon for working with data streams. * daemon/Makefile.am: Add stream.c/stream.h to build * daemon/stream.c, qemud/stream.h: Generic helper functions for creating new streams, associating streams with clients, finding existing streams for a client and removing/deleting streams. * src/remote/remote_protocol.x: Add a new 'REMOTE_STREAM' constant for the 'enum remote_message_type' for encoding stream data in wire messages. Add a new 'REMOTE_CONTINUE' constant to 'enum remote_message_status' to indicate further data stream messsages are expected to follow. Document how the remote_message_header is used to encode data streams * src/remote/remote_protocol.h: Regenerate * daemon/dispatch.c: Remove assumption that a error message sent to client is always type=REMOTE_REPLY. It may now also be type=REMOTE_STREAM. Add convenient method for sending outgoing stream data packets. Log and ignore non-filtered incoming stream packets. Add a method for serializing a stream error message * daemon/dispatch.h: Add API for serializing stream errors and sending stream data packets * daemon/qemud.h: Add struct qemud_client_stream for tracking active data streams for clients. Tweak filter function operation so that it accepts a client object too. * daemon/qemud.c: Refactor code for free'ing message objects which have been fully transmitted into separate method. Release all active streams when client shuts down. Change filter function to be responsible for queueing the message
2009-07-10 12:06:36 +00:00
}
prev = curr;
curr = curr->next;
}
return -1;
}
void
daemonRemoveAllClientStreams(daemonClientStream *stream)
{
daemonClientStream *tmp;
VIR_DEBUG("stream=%p", stream);
while (stream) {
tmp = stream->next;
if (!stream->closed) {
stream->closed = true;
virStreamEventRemoveCallback(stream->st);
virStreamAbort(stream->st);
}
daemonFreeClientStream(NULL, stream);
VIR_DEBUG("next stream=%p", tmp);
stream = tmp;
}
}
/*
* Returns:
* -1 if fatal error occurred
* 0 if message was fully processed
* 1 if message is still being processed
*/
static int
daemonStreamHandleWriteData(virNetServerClientPtr client,
daemonClientStream *stream,
virNetMessagePtr msg)
{
int ret;
VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u, len=%zu, offset=%zu",
client, stream, msg->header.proc, msg->header.serial,
msg->bufferLength, msg->bufferOffset);
ret = virStreamSend(stream->st,
msg->buffer + msg->bufferOffset,
msg->bufferLength - msg->bufferOffset);
if (ret > 0) {
msg->bufferOffset += ret;
/* Partial write, so indicate we have more todo later */
if (msg->bufferOffset < msg->bufferLength)
return 1;
} else if (ret == -2) {
/* Blocking, so indicate we have more todo later */
return 1;
} else {
virNetMessageError rerr;
memset(&rerr, 0, sizeof(rerr));
VIR_INFO("Stream send failed");
stream->closed = true;
virStreamEventRemoveCallback(stream->st);
virStreamAbort(stream->st);
return virNetServerProgramSendReplyError(stream->prog,
client,
msg,
&rerr,
&msg->header);
}
return 0;
}
/*
* Process a finish handshake from the client.
*
* Returns a VIR_NET_OK confirmation if successful, or a VIR_NET_ERROR
* if there was a stream error
*
* Returns 0 if successfully sent RPC reply, -1 upon fatal error
*/
static int
daemonStreamHandleFinish(virNetServerClientPtr client,
daemonClientStream *stream,
virNetMessagePtr msg)
{
int ret;
VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u",
client, stream, msg->header.proc, msg->header.serial);
stream->closed = true;
virStreamEventRemoveCallback(stream->st);
ret = virStreamFinish(stream->st);
if (ret < 0) {
virNetMessageError rerr;
memset(&rerr, 0, sizeof(rerr));
return virNetServerProgramSendReplyError(stream->prog,
client,
msg,
&rerr,
&msg->header);
} else {
/* Send zero-length confirm */
return virNetServerProgramSendStreamData(stream->prog,
client,
msg,
stream->procedure,
stream->serial,
NULL, 0);
}
}
/*
* Process an abort request from the client.
*
* Returns 0 if successfully aborted, -1 upon error
*/
static int
daemonStreamHandleAbort(virNetServerClientPtr client,
daemonClientStream *stream,
virNetMessagePtr msg)
{
VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u",
client, stream, msg->header.proc, msg->header.serial);
int ret;
bool raise_error = false;
stream->closed = true;
virStreamEventRemoveCallback(stream->st);
ret = virStreamAbort(stream->st);
maint: use consistent if-else braces in remaining spots I'm about to add a syntax check that enforces our documented HACKING style of always using matching {} on if-else statements. This patch focuses on all remaining problems, where there weren't enough issues to warrant splitting it further. * src/remote/remote_driver.c (doRemoteOpen): Correct use of {}. * src/security/virt-aa-helper.c (vah_add_path, valid_path, main): Likewise. * src/rpc/virnetsocket.c (virNetSocketNewConnectLibSSH2): Likewise. * src/esx/esx_vi_types.c (esxVI_Type_FromString): Likewise. * src/uml/uml_driver.c (umlDomainDetachDevice): Likewise. * src/util/viralloc.c (virShrinkN): Likewise. * src/util/virbuffer.c (virBufferURIEncodeString): Likewise. * src/util/virdbus.c (virDBusCall): Likewise. * src/util/virnetdev.c (virNetDevValidateConfig): Likewise. * src/util/virnetdevvportprofile.c (virNetDevVPortProfileGetNthParent): Likewise. * src/util/virpci.c (virPCIDeviceIterDevices) (virPCIDeviceWaitForCleanup) (virPCIDeviceIsBehindSwitchLackingACS): Likewise. * src/util/virsocketaddr.c (virSocketAddrGetNumNetmaskBits): Likewise. * src/util/viruri.c (virURIParseParams): Likewise. * daemon/stream.c (daemonStreamHandleAbort): Likewise. * tests/testutils.c (virtTestResult): Likewise. * tests/cputest.c (cpuTestBaseline): Likewise. * tools/virsh-domain.c (cmdDomPMSuspend): Likewise. * tools/virsh-host.c (cmdNodeSuspend): Likewise. * src/esx/esx_vi_generator.py (Type.generate_typefromstring): Tweak generated code. Signed-off-by: Eric Blake <eblake@redhat.com>
2014-09-03 19:39:21 +00:00
if (msg->header.status == VIR_NET_ERROR) {
VIR_INFO("stream aborted at client request");
raise_error = (ret < 0);
maint: use consistent if-else braces in remaining spots I'm about to add a syntax check that enforces our documented HACKING style of always using matching {} on if-else statements. This patch focuses on all remaining problems, where there weren't enough issues to warrant splitting it further. * src/remote/remote_driver.c (doRemoteOpen): Correct use of {}. * src/security/virt-aa-helper.c (vah_add_path, valid_path, main): Likewise. * src/rpc/virnetsocket.c (virNetSocketNewConnectLibSSH2): Likewise. * src/esx/esx_vi_types.c (esxVI_Type_FromString): Likewise. * src/uml/uml_driver.c (umlDomainDetachDevice): Likewise. * src/util/viralloc.c (virShrinkN): Likewise. * src/util/virbuffer.c (virBufferURIEncodeString): Likewise. * src/util/virdbus.c (virDBusCall): Likewise. * src/util/virnetdev.c (virNetDevValidateConfig): Likewise. * src/util/virnetdevvportprofile.c (virNetDevVPortProfileGetNthParent): Likewise. * src/util/virpci.c (virPCIDeviceIterDevices) (virPCIDeviceWaitForCleanup) (virPCIDeviceIsBehindSwitchLackingACS): Likewise. * src/util/virsocketaddr.c (virSocketAddrGetNumNetmaskBits): Likewise. * src/util/viruri.c (virURIParseParams): Likewise. * daemon/stream.c (daemonStreamHandleAbort): Likewise. * tests/testutils.c (virtTestResult): Likewise. * tests/cputest.c (cpuTestBaseline): Likewise. * tools/virsh-domain.c (cmdDomPMSuspend): Likewise. * tools/virsh-host.c (cmdNodeSuspend): Likewise. * src/esx/esx_vi_generator.py (Type.generate_typefromstring): Tweak generated code. Signed-off-by: Eric Blake <eblake@redhat.com>
2014-09-03 19:39:21 +00:00
} else {
virReportError(VIR_ERR_RPC,
_("stream aborted with unexpected status %d"),
msg->header.status);
raise_error = true;
}
if (raise_error) {
virNetMessageError rerr;
memset(&rerr, 0, sizeof(rerr));
return virNetServerProgramSendReplyError(remoteProgram,
client,
msg,
&rerr,
&msg->header);
} else {
/* Send zero-length confirm */
return virNetServerProgramSendStreamData(stream->prog,
client,
msg,
stream->procedure,
stream->serial,
NULL, 0);
}
}
/*
* Called when the stream is signalled has being able to accept
* data writes. Will process all pending incoming messages
* until they're all gone, or I/O blocks
*
* Returns 0 on success, or -1 upon fatal error
*/
static int
daemonStreamHandleWrite(virNetServerClientPtr client,
daemonClientStream *stream)
{
VIR_DEBUG("client=%p, stream=%p", client, stream);
while (stream->rx && !stream->closed) {
virNetMessagePtr msg = stream->rx;
int ret;
switch (msg->header.status) {
case VIR_NET_OK:
ret = daemonStreamHandleFinish(client, stream, msg);
break;
case VIR_NET_CONTINUE:
ret = daemonStreamHandleWriteData(client, stream, msg);
break;
case VIR_NET_ERROR:
default:
ret = daemonStreamHandleAbort(client, stream, msg);
break;
}
if (ret > 0)
break; /* still processing data from msg */
virNetMessageQueueServe(&stream->rx);
if (ret < 0) {
virNetMessageFree(msg);
virNetServerClientImmediateClose(client);
return -1;
}
/* 'CONTINUE' messages don't send a reply (unless error
* occurred), so to release the 'msg' object we need to
* send a fake zero-length reply. Nothing actually gets
* onto the wire, but this causes the client to reset
* its active request count / throttling
*/
if (msg->header.status == VIR_NET_CONTINUE) {
2011-08-31 16:42:58 +00:00
virNetMessageClear(msg);
msg->header.type = VIR_NET_REPLY;
if (virNetServerClientSendMessage(client, msg) < 0) {
virNetMessageFree(msg);
virNetServerClientImmediateClose(client);
return -1;
}
}
}
return 0;
}
/*
* Invoked when a stream is signalled as having data
2012-10-11 16:31:20 +00:00
* available to read. This reads up to one message
* worth of data, and then queues that for transmission
* to the client.
*
2016-12-01 20:50:08 +00:00
* Returns 0 if data was queued for TX, or an error RPC
* was sent, or -1 on fatal error, indicating client should
* be killed
*/
static int
daemonStreamHandleRead(virNetServerClientPtr client,
daemonClientStream *stream)
{
virNetMessagePtr msg = NULL;
virNetMessageError rerr;
char *buffer;
size_t bufferLen = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX;
int ret = -1;
int rv;
VIR_DEBUG("client=%p, stream=%p tx=%d closed=%d",
client, stream, stream->tx, stream->closed);
/* We might have had an event pending before we shut
* down the stream, so if we're marked as closed,
* then do nothing
*/
if (stream->closed)
return 0;
/* Shouldn't ever be called unless we're marked able to
* transmit, but doesn't hurt to check */
if (!stream->tx)
return 0;
memset(&rerr, 0, sizeof(rerr));
if (VIR_ALLOC_N(buffer, bufferLen) < 0)
return -1;
if (!(msg = virNetMessageNew(false)))
goto cleanup;
rv = virStreamRecv(stream->st, buffer, bufferLen);
if (rv == -2) {
/* Should never get this, since we're only called when we know
* we're readable, but hey things change... */
} else if (rv < 0) {
if (virNetServerProgramSendStreamError(remoteProgram,
client,
msg,
&rerr,
stream->procedure,
stream->serial) < 0)
goto cleanup;
msg = NULL;
} else {
stream->tx = false;
if (rv == 0)
stream->recvEOF = true;
msg->cb = daemonStreamMessageFinished;
msg->opaque = stream;
stream->refs++;
if (virNetServerProgramSendStreamData(remoteProgram,
client,
msg,
stream->procedure,
stream->serial,
buffer, rv) < 0)
goto cleanup;
msg = NULL;
}
ret = 0;
cleanup:
VIR_FREE(buffer);
virNetMessageFree(msg);
return ret;
}