*/
#include "qemu/osdep.h"
-#include "qemu-common.h"
#include "qemu/error-report.h"
#include "trace.h"
#include "qapi/error.h"
#include "migration/migration.h"
#include "util.h"
+#include "block/aio-wait.h"
+#include "qemu/coroutine.h"
+
#define TYPE_COLO_COMPARE "colo-compare"
-#define COLO_COMPARE(obj) \
- OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
+typedef struct CompareState CompareState;
+DECLARE_INSTANCE_CHECKER(CompareState, COLO_COMPARE,
+ TYPE_COLO_COMPARE)
static QTAILQ_HEAD(, CompareState) net_compares =
QTAILQ_HEAD_INITIALIZER(net_compares);
#define COLO_COMPARE_FREE_PRIMARY 0x01
#define COLO_COMPARE_FREE_SECONDARY 0x02
-#define REGULAR_PACKET_CHECK_MS 3000
+#define REGULAR_PACKET_CHECK_MS 1000
#define DEFAULT_TIME_OUT_MS 3000
+/* #define DEBUG_COLO_PACKETS */
+
+static QemuMutex colo_compare_mutex;
+static bool colo_compare_active;
static QemuMutex event_mtx;
static QemuCond event_complete_cond;
static int event_unhandled_count;
+static uint32_t max_queue_size;
/*
* + CompareState ++
* |packet | |packet + |packet | |packet +
* +--------+ +--------+ +--------+ +--------+
*/
-typedef struct CompareState {
+
+typedef struct SendCo {
+ Coroutine *co;
+ struct CompareState *s;
+ CharBackend *chr;
+ GQueue send_list;
+ bool notify_remote_frame;
+ bool done;
+ int ret;
+} SendCo;
+
+typedef struct SendEntry {
+ uint32_t size;
+ uint32_t vnet_hdr_len;
+ uint8_t *buf;
+} SendEntry;
+
+struct CompareState {
Object parent;
char *pri_indev;
SocketReadState pri_rs;
SocketReadState sec_rs;
SocketReadState notify_rs;
+ SendCo out_sendco;
+ SendCo notify_sendco;
bool vnet_hdr;
- uint32_t compare_timeout;
+ uint64_t compare_timeout;
uint32_t expired_scan_cycle;
/*
enum colo_event event;
QTAILQ_ENTRY(CompareState) next;
-} CompareState;
+};
typedef struct CompareClass {
ObjectClass parent_class;
SECONDARY_IN,
};
+static const char *colo_mode[] = {
+ [PRIMARY_IN] = "primary",
+ [SECONDARY_IN] = "secondary",
+};
static int compare_chr_send(CompareState *s,
- const uint8_t *buf,
+ uint8_t *buf,
uint32_t size,
uint32_t vnet_hdr_len,
- bool notify_remote_frame);
+ bool notify_remote_frame,
+ bool zero_copy);
static bool packet_matches_str(const char *str,
const uint8_t *buf,
return false;
}
- return !memcmp(str, buf, strlen(str));
+ return !memcmp(str, buf, packet_len);
}
static void notify_remote_frame(CompareState *s)
char msg[] = "DO_CHECKPOINT";
int ret = 0;
- ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true);
+ ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
if (ret < 0) {
error_report("Notify Xen COLO-frame failed");
}
}
}
+/* Use restricted to colo_insert_packet() */
static gint seq_sorter(Packet *a, Packet *b, gpointer data)
{
- struct tcp_hdr *atcp, *btcp;
-
- atcp = (struct tcp_hdr *)(a->transport_header);
- btcp = (struct tcp_hdr *)(b->transport_header);
- return ntohl(atcp->th_seq) - ntohl(btcp->th_seq);
+ return b->tcp_seq - a->tcp_seq;
}
static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
pkt->tcp_seq = ntohl(tcphd->th_seq);
pkt->tcp_ack = ntohl(tcphd->th_ack);
- *max_ack = *max_ack > pkt->tcp_ack ? *max_ack : pkt->tcp_ack;
+ /* Need to consider ACK will bigger than uint32_t MAX */
+ *max_ack = pkt->tcp_ack - *max_ack > 0 ? pkt->tcp_ack : *max_ack;
pkt->header_size = pkt->transport_header - (uint8_t *)pkt->data
- + (tcphd->th_off << 2) - pkt->vnet_hdr_len;
+ + (tcphd->th_off << 2);
pkt->payload_size = pkt->size - pkt->header_size;
pkt->seq_end = pkt->tcp_seq + pkt->payload_size;
pkt->flags = tcphd->th_flags;
*/
static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
{
- if (g_queue_get_length(queue) <= MAX_QUEUE_SIZE) {
+ if (g_queue_get_length(queue) <= max_queue_size) {
if (pkt->ip->ip_p == IPPROTO_TCP) {
fill_pkt_tcp_info(pkt, max_ack);
g_queue_insert_sorted(queue,
ConnectionKey key;
Packet *pkt = NULL;
Connection *conn;
+ int ret;
if (mode == PRIMARY_IN) {
pkt = packet_new(s->pri_rs.buf,
pkt = NULL;
return -1;
}
- fill_connection_key(pkt, &key);
+ fill_connection_key(pkt, &key, false);
conn = connection_get(s->connection_track_table,
&key,
}
if (mode == PRIMARY_IN) {
- if (!colo_insert_packet(&conn->primary_list, pkt, &conn->pack)) {
- error_report("colo compare primary queue size too big,"
- "drop packet");
- }
+ ret = colo_insert_packet(&conn->primary_list, pkt, &conn->pack);
} else {
- if (!colo_insert_packet(&conn->secondary_list, pkt, &conn->sack)) {
- error_report("colo compare secondary queue size too big,"
- "drop packet");
- }
+ ret = colo_insert_packet(&conn->secondary_list, pkt, &conn->sack);
+ }
+
+ if (!ret) {
+ trace_colo_compare_drop_packet(colo_mode[mode],
+ "queue size too big, drop packet");
+ packet_destroy(pkt, NULL);
+ pkt = NULL;
}
+
*con = conn;
return 0;
pkt->data,
pkt->size,
pkt->vnet_hdr_len,
- false);
+ false,
+ true);
if (ret < 0) {
error_report("colo send primary packet failed");
}
trace_colo_compare_main("packet same and release packet");
- packet_destroy(pkt, NULL);
+ packet_destroy_partial(pkt, NULL);
}
/*
uint16_t len)
{
- if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
+ if (trace_event_get_state_backends(TRACE_COLO_COMPARE_IP_INFO)) {
char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
* can ensure that the packet's payload is acknowledged by
* primary and secondary.
*/
- uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
+ uint32_t min_ack = conn->pack - conn->sack > 0 ?
+ conn->sack : conn->pack;
pri:
if (g_queue_is_empty(&conn->primary_list)) {
return;
}
- ppkt = g_queue_pop_head(&conn->primary_list);
+ ppkt = g_queue_pop_tail(&conn->primary_list);
sec:
if (g_queue_is_empty(&conn->secondary_list)) {
- g_queue_push_head(&conn->primary_list, ppkt);
+ g_queue_push_tail(&conn->primary_list, ppkt);
return;
}
- spkt = g_queue_pop_head(&conn->secondary_list);
+ spkt = g_queue_pop_tail(&conn->secondary_list);
if (ppkt->tcp_seq == ppkt->seq_end) {
colo_release_primary_pkt(s, ppkt);
}
}
if (!ppkt) {
- g_queue_push_head(&conn->secondary_list, spkt);
+ g_queue_push_tail(&conn->secondary_list, spkt);
goto pri;
}
}
if (mark == COLO_COMPARE_FREE_PRIMARY) {
conn->compare_seq = ppkt->seq_end;
colo_release_primary_pkt(s, ppkt);
- g_queue_push_head(&conn->secondary_list, spkt);
+ g_queue_push_tail(&conn->secondary_list, spkt);
goto pri;
- }
- if (mark == COLO_COMPARE_FREE_SECONDARY) {
+ } else if (mark == COLO_COMPARE_FREE_SECONDARY) {
conn->compare_seq = spkt->seq_end;
packet_destroy(spkt, NULL);
goto sec;
- }
- if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
+ } else if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
conn->compare_seq = ppkt->seq_end;
colo_release_primary_pkt(s, ppkt);
packet_destroy(spkt, NULL);
goto pri;
}
} else {
- g_queue_push_head(&conn->primary_list, ppkt);
- g_queue_push_head(&conn->secondary_list, spkt);
+ g_queue_push_tail(&conn->primary_list, ppkt);
+ g_queue_push_tail(&conn->secondary_list, spkt);
- qemu_hexdump((char *)ppkt->data, stderr,
- "colo-compare ppkt", ppkt->size);
- qemu_hexdump((char *)spkt->data, stderr,
- "colo-compare spkt", spkt->size);
+#ifdef DEBUG_COLO_PACKETS
+ qemu_hexdump(stderr, "colo-compare ppkt", ppkt->data, ppkt->size);
+ qemu_hexdump(stderr, "colo-compare spkt", spkt->data, spkt->size);
+#endif
colo_compare_inconsistency_notify(s);
}
ppkt->size - offset)) {
trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
- if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
- qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
- ppkt->size);
- qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
- spkt->size);
- }
+#ifdef DEBUG_COLO_PACKETS
+ qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
+ qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
+#endif
return -1;
} else {
return 0;
ppkt->size);
trace_colo_compare_icmp_miscompare("Secondary pkt size",
spkt->size);
- if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
- qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
- ppkt->size);
- qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
- spkt->size);
- }
+#ifdef DEBUG_COLO_PACKETS
+ qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
+ qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
+#endif
return -1;
} else {
return 0;
uint16_t offset = ppkt->vnet_hdr_len;
trace_colo_compare_main("compare other");
- if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
- char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
-
- strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
- strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
- strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
- strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
-
- trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
- pri_ip_dst, spkt->size,
- sec_ip_src, sec_ip_dst);
- }
-
if (ppkt->size != spkt->size) {
trace_colo_compare_main("Other: payload size of packets are different");
return -1;
static int colo_old_packet_check_one_conn(Connection *conn,
CompareState *s)
{
- GList *result = NULL;
-
- result = g_queue_find_custom(&conn->primary_list,
- &s->compare_timeout,
- (GCompareFunc)colo_old_packet_check_one);
+ if (!g_queue_is_empty(&conn->primary_list)) {
+ if (g_queue_find_custom(&conn->primary_list,
+ &s->compare_timeout,
+ (GCompareFunc)colo_old_packet_check_one))
+ goto out;
+ }
- if (result) {
- /* Do checkpoint will flush old packet */
- colo_compare_inconsistency_notify(s);
- return 0;
+ if (!g_queue_is_empty(&conn->secondary_list)) {
+ if (g_queue_find_custom(&conn->secondary_list,
+ &s->compare_timeout,
+ (GCompareFunc)colo_old_packet_check_one))
+ goto out;
}
return 1;
+
+out:
+ /* Do checkpoint will flush old packet */
+ colo_compare_inconsistency_notify(s);
+ return 0;
}
/*
while (!g_queue_is_empty(&conn->primary_list) &&
!g_queue_is_empty(&conn->secondary_list)) {
- pkt = g_queue_pop_head(&conn->primary_list);
+ pkt = g_queue_pop_tail(&conn->primary_list);
result = g_queue_find_custom(&conn->secondary_list,
pkt, (GCompareFunc)HandlePacket);
if (result) {
colo_release_primary_pkt(s, pkt);
- g_queue_remove(&conn->secondary_list, result->data);
+ packet_destroy(result->data, NULL);
+ g_queue_delete_link(&conn->secondary_list, result);
} else {
/*
* If one packet arrive late, the secondary_list or
* timeout, it will trigger a checkpoint request.
*/
trace_colo_compare_main("packet different");
- g_queue_push_head(&conn->primary_list, pkt);
+ g_queue_push_tail(&conn->primary_list, pkt);
colo_compare_inconsistency_notify(s);
break;
}
}
-static int compare_chr_send(CompareState *s,
- const uint8_t *buf,
- uint32_t size,
- uint32_t vnet_hdr_len,
- bool notify_remote_frame)
+static void coroutine_fn _compare_chr_send(void *opaque)
{
+ SendCo *sendco = opaque;
+ CompareState *s = sendco->s;
int ret = 0;
- uint32_t len = htonl(size);
- if (!size) {
- return 0;
- }
+ while (!g_queue_is_empty(&sendco->send_list)) {
+ SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
+ uint32_t len = htonl(entry->size);
- if (notify_remote_frame) {
- ret = qemu_chr_fe_write_all(&s->chr_notify_dev,
- (uint8_t *)&len,
- sizeof(len));
- } else {
- ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)&len, sizeof(len));
- }
+ ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
- if (ret != sizeof(len)) {
- goto err;
- }
+ if (ret != sizeof(len)) {
+ g_free(entry->buf);
+ g_slice_free(SendEntry, entry);
+ goto err;
+ }
- if (s->vnet_hdr) {
- /*
- * We send vnet header len make other module(like filter-redirector)
- * know how to parse net packet correctly.
- */
- len = htonl(vnet_hdr_len);
+ if (!sendco->notify_remote_frame && s->vnet_hdr) {
+ /*
+ * We send vnet header len make other module(like filter-redirector)
+ * know how to parse net packet correctly.
+ */
+ len = htonl(entry->vnet_hdr_len);
- if (!notify_remote_frame) {
- ret = qemu_chr_fe_write_all(&s->chr_out,
+ ret = qemu_chr_fe_write_all(sendco->chr,
(uint8_t *)&len,
sizeof(len));
+
+ if (ret != sizeof(len)) {
+ g_free(entry->buf);
+ g_slice_free(SendEntry, entry);
+ goto err;
+ }
}
- if (ret != sizeof(len)) {
+ ret = qemu_chr_fe_write_all(sendco->chr,
+ (uint8_t *)entry->buf,
+ entry->size);
+
+ if (ret != entry->size) {
+ g_free(entry->buf);
+ g_slice_free(SendEntry, entry);
goto err;
}
+
+ g_free(entry->buf);
+ g_slice_free(SendEntry, entry);
+ }
+
+ sendco->ret = 0;
+ goto out;
+
+err:
+ while (!g_queue_is_empty(&sendco->send_list)) {
+ SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
+ g_free(entry->buf);
+ g_slice_free(SendEntry, entry);
}
+ sendco->ret = ret < 0 ? ret : -EIO;
+out:
+ sendco->co = NULL;
+ sendco->done = true;
+ aio_wait_kick();
+}
+
+static int compare_chr_send(CompareState *s,
+ uint8_t *buf,
+ uint32_t size,
+ uint32_t vnet_hdr_len,
+ bool notify_remote_frame,
+ bool zero_copy)
+{
+ SendCo *sendco;
+ SendEntry *entry;
if (notify_remote_frame) {
- ret = qemu_chr_fe_write_all(&s->chr_notify_dev,
- (uint8_t *)buf,
- size);
+ sendco = &s->notify_sendco;
} else {
- ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)buf, size);
+ sendco = &s->out_sendco;
}
- if (ret != size) {
- goto err;
+ if (!size) {
+ return -1;
}
- return 0;
+ entry = g_slice_new(SendEntry);
+ entry->size = size;
+ entry->vnet_hdr_len = vnet_hdr_len;
+ if (zero_copy) {
+ entry->buf = buf;
+ } else {
+ entry->buf = g_malloc(size);
+ memcpy(entry->buf, buf, size);
+ }
+ g_queue_push_tail(&sendco->send_list, entry);
+
+ if (sendco->done) {
+ sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
+ sendco->done = false;
+ qemu_coroutine_enter(sendco->co);
+ if (sendco->done) {
+ /* report early errors */
+ return sendco->ret;
+ }
+ }
-err:
- return ret < 0 ? ret : -EIO;
+ /* assume success */
+ return 0;
}
static int compare_chr_can_read(void *opaque)
/* if have old packet we will notify checkpoint */
colo_old_packet_check(s);
- timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
+ timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
s->expired_scan_cycle);
}
void colo_notify_compares_event(void *opaque, int event, Error **errp)
{
CompareState *s;
+ qemu_mutex_lock(&colo_compare_mutex);
+
+ if (!colo_compare_active) {
+ qemu_mutex_unlock(&colo_compare_mutex);
+ return;
+ }
qemu_mutex_lock(&event_mtx);
QTAILQ_FOREACH(s, &net_compares, next) {
}
qemu_mutex_unlock(&event_mtx);
+ qemu_mutex_unlock(&colo_compare_mutex);
}
static void colo_compare_timer_init(CompareState *s)
{
AioContext *ctx = iothread_get_aio_context(s->iothread);
- s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
+ s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_HOST,
SCALE_MS, check_old_packet_regular,
s);
- timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
+ timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
s->expired_scan_cycle);
}
static void colo_compare_timer_del(CompareState *s)
{
if (s->packet_check_timer) {
- timer_del(s->packet_check_timer);
timer_free(s->packet_check_timer);
s->packet_check_timer = NULL;
}
static void colo_compare_iothread(CompareState *s)
{
+ AioContext *ctx = iothread_get_aio_context(s->iothread);
object_ref(OBJECT(s->iothread));
s->worker_context = iothread_get_g_main_context(s->iothread);
}
colo_compare_timer_init(s);
- s->event_bh = qemu_bh_new(colo_compare_handle_event, s);
+ s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
}
static char *compare_get_pri_indev(Object *obj, Error **errp)
Error **errp)
{
CompareState *s = COLO_COMPARE(obj);
- uint32_t value = s->compare_timeout;
+ uint64_t value = s->compare_timeout;
- visit_type_uint32(v, name, &value, errp);
+ visit_type_uint64(v, name, &value, errp);
}
static void compare_set_timeout(Object *obj, Visitor *v,
Error **errp)
{
CompareState *s = COLO_COMPARE(obj);
- Error *local_err = NULL;
uint32_t value;
- visit_type_uint32(v, name, &value, &local_err);
- if (local_err) {
- goto out;
+ if (!visit_type_uint32(v, name, &value, errp)) {
+ return;
}
if (!value) {
- error_setg(&local_err, "Property '%s.%s' requires a positive value",
+ error_setg(errp, "Property '%s.%s' requires a positive value",
object_get_typename(obj), name);
- goto out;
+ return;
}
s->compare_timeout = value;
-
-out:
- error_propagate(errp, local_err);
}
static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
Error **errp)
{
CompareState *s = COLO_COMPARE(obj);
- Error *local_err = NULL;
uint32_t value;
- visit_type_uint32(v, name, &value, &local_err);
+ if (!visit_type_uint32(v, name, &value, errp)) {
+ return;
+ }
+ if (!value) {
+ error_setg(errp, "Property '%s.%s' requires a positive value",
+ object_get_typename(obj), name);
+ return;
+ }
+ s->expired_scan_cycle = value;
+}
+
+static void get_max_queue_size(Object *obj, Visitor *v,
+ const char *name, void *opaque,
+ Error **errp)
+{
+ uint32_t value = max_queue_size;
+
+ visit_type_uint32(v, name, &value, errp);
+}
+
+static void set_max_queue_size(Object *obj, Visitor *v,
+ const char *name, void *opaque,
+ Error **errp)
+{
+ Error *local_err = NULL;
+ uint64_t value;
+
+ visit_type_uint64(v, name, &value, &local_err);
if (local_err) {
goto out;
}
object_get_typename(obj), name);
goto out;
}
- s->expired_scan_cycle = value;
+ max_queue_size = value;
out:
error_propagate(errp, local_err);
pri_rs->buf,
pri_rs->packet_len,
pri_rs->vnet_hdr_len,
+ false,
false);
} else {
/* compare packet in the specified connection */
if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
notify_rs->buf,
notify_rs->packet_len)) {
- ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true);
+ ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
if (ret < 0) {
error_report("Notify Xen COLO-frame INIT failed");
}
}
if (!s->expired_scan_cycle) {
- /* Set default value to 3000 MS */
+ /* Set default value to 1000 MS */
s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
}
+ if (!max_queue_size) {
+ /* Set default queue size to 1024 */
+ max_queue_size = MAX_QUEUE_SIZE;
+ }
+
if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
!qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
return;
s->vnet_hdr);
}
- QTAILQ_INSERT_TAIL(&net_compares, s, next);
+ s->out_sendco.s = s;
+ s->out_sendco.chr = &s->chr_out;
+ s->out_sendco.notify_remote_frame = false;
+ s->out_sendco.done = true;
+ g_queue_init(&s->out_sendco.send_list);
- g_queue_init(&s->conn_list);
+ if (s->notify_dev) {
+ s->notify_sendco.s = s;
+ s->notify_sendco.chr = &s->chr_notify_dev;
+ s->notify_sendco.notify_remote_frame = true;
+ s->notify_sendco.done = true;
+ g_queue_init(&s->notify_sendco.send_list);
+ }
- qemu_mutex_init(&event_mtx);
- qemu_cond_init(&event_complete_cond);
+ g_queue_init(&s->conn_list);
s->connection_track_table = g_hash_table_new_full(connection_key_hash,
connection_key_equal,
connection_destroy);
colo_compare_iothread(s);
+
+ qemu_mutex_lock(&colo_compare_mutex);
+ if (!colo_compare_active) {
+ qemu_mutex_init(&event_mtx);
+ qemu_cond_init(&event_complete_cond);
+ colo_compare_active = true;
+ }
+ QTAILQ_INSERT_TAIL(&net_compares, s, next);
+ qemu_mutex_unlock(&colo_compare_mutex);
+
return;
}
Packet *pkt = NULL;
while (!g_queue_is_empty(&conn->primary_list)) {
- pkt = g_queue_pop_head(&conn->primary_list);
+ pkt = g_queue_pop_tail(&conn->primary_list);
compare_chr_send(s,
pkt->data,
pkt->size,
pkt->vnet_hdr_len,
- false);
- packet_destroy(pkt, NULL);
+ false,
+ true);
+ packet_destroy_partial(pkt, NULL);
}
while (!g_queue_is_empty(&conn->secondary_list)) {
- pkt = g_queue_pop_head(&conn->secondary_list);
+ pkt = g_queue_pop_tail(&conn->secondary_list);
packet_destroy(pkt, NULL);
}
}
object_property_add_str(obj, "notify_dev",
compare_get_notify_dev, compare_set_notify_dev);
- object_property_add(obj, "compare_timeout", "uint32",
+ object_property_add(obj, "compare_timeout", "uint64",
compare_get_timeout,
compare_set_timeout, NULL, NULL);
compare_get_expired_scan_cycle,
compare_set_expired_scan_cycle, NULL, NULL);
+ object_property_add(obj, "max_queue_size", "uint32",
+ get_max_queue_size,
+ set_max_queue_size, NULL, NULL);
+
s->vnet_hdr = false;
object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
compare_set_vnet_hdr);
}
+void colo_compare_cleanup(void)
+{
+ CompareState *tmp = NULL;
+ CompareState *n = NULL;
+
+ QTAILQ_FOREACH_SAFE(tmp, &net_compares, next, n) {
+ object_unparent(OBJECT(tmp));
+ }
+}
+
static void colo_compare_finalize(Object *obj)
{
CompareState *s = COLO_COMPARE(obj);
CompareState *tmp = NULL;
+ qemu_mutex_lock(&colo_compare_mutex);
+ QTAILQ_FOREACH(tmp, &net_compares, next) {
+ if (tmp == s) {
+ QTAILQ_REMOVE(&net_compares, s, next);
+ break;
+ }
+ }
+ if (QTAILQ_EMPTY(&net_compares)) {
+ colo_compare_active = false;
+ qemu_mutex_destroy(&event_mtx);
+ qemu_cond_destroy(&event_complete_cond);
+ }
+ qemu_mutex_unlock(&colo_compare_mutex);
+
qemu_chr_fe_deinit(&s->chr_pri_in, false);
qemu_chr_fe_deinit(&s->chr_sec_in, false);
qemu_chr_fe_deinit(&s->chr_out, false);
qemu_chr_fe_deinit(&s->chr_notify_dev, false);
}
- if (s->iothread) {
- colo_compare_timer_del(s);
- }
+ colo_compare_timer_del(s);
qemu_bh_delete(s->event_bh);
- QTAILQ_FOREACH(tmp, &net_compares, next) {
- if (tmp == s) {
- QTAILQ_REMOVE(&net_compares, s, next);
- break;
- }
+ AioContext *ctx = iothread_get_aio_context(s->iothread);
+ aio_context_acquire(ctx);
+ AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
+ if (s->notify_dev) {
+ AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
}
+ aio_context_release(ctx);
/* Release all unhandled packets after compare thead exited */
g_queue_foreach(&s->conn_list, colo_flush_packets, s);
+ AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
g_queue_clear(&s->conn_list);
+ g_queue_clear(&s->out_sendco.send_list);
+ if (s->notify_dev) {
+ g_queue_clear(&s->notify_sendco.send_list);
+ }
if (s->connection_track_table) {
g_hash_table_destroy(s->connection_track_table);
}
- if (s->iothread) {
- object_unref(OBJECT(s->iothread));
- }
-
- qemu_mutex_destroy(&event_mtx);
- qemu_cond_destroy(&event_complete_cond);
+ object_unref(OBJECT(s->iothread));
g_free(s->pri_indev);
g_free(s->sec_indev);
g_free(s->notify_dev);
}
+static void __attribute__((__constructor__)) colo_compare_init_globals(void)
+{
+ colo_compare_active = false;
+ qemu_mutex_init(&colo_compare_mutex);
+}
+
static const TypeInfo colo_compare_info = {
.name = TYPE_COLO_COMPARE,
.parent = TYPE_OBJECT,