rdma.c · samitouri/QOSamiQemu · siGit Code & Deploy

Code Commits Issues Pulls Sessions CI/CD

master

c 3,869 lines 115 KB

Raw

1	/*
2	* RDMA protocol and interfaces
3	*
4	* Copyright IBM, Corp. 2010-2013
5	* Copyright Red Hat, Inc. 2015-2016
6	*
7	* Authors:
8	* Michael R. Hines <mrhines@us.ibm.com>
9	* Jiuxing Liu <jl@us.ibm.com>
10	* Daniel P. Berrange <berrange@redhat.com>
11	*
12	* This work is licensed under the terms of the GNU GPL, version 2 or
13	* later. See the COPYING file in the top-level directory.
14	*
15	*/
16
17	#include "qemu/osdep.h"
18	#include "channel.h"
19	#include "qapi/error.h"
20	#include "qemu/cutils.h"
21	#include "channel.h"
22	#include "exec/target_page.h"
23	#include "rdma.h"
24	#include "migration.h"
25	#include "migration-stats.h"
26	#include "qemu-file.h"
27	#include "ram.h"
28	#include "qemu/error-report.h"
29	#include "qemu/main-loop.h"
30	#include "qemu/module.h"
31	#include "qemu/rcu.h"
32	#include "qemu/sockets.h"
33	#include "qemu/bitmap.h"
34	#include "qemu/coroutine.h"
35	#include "system/memory.h"
36	#include <sys/socket.h>
37	#include <netdb.h>
38	#include <arpa/inet.h>
39	#include <rdma/rdma_cma.h>
40	#include "trace.h"
41	#include "qom/object.h"
42	#include "options.h"
43	#include <poll.h>
44
45	#define RDMA_RESOLVE_TIMEOUT_MS 10000
46
47	/* Do not merge data if larger than this. */
48	static inline uint64_t rdma_merge_max(void)
49	{
50	return migrate_rdma_chunk_size() * 2;
51	}
52
53	#define RDMA_SIGNALED_SEND_MAX 512
54
55	/*
56	* This is only for non-live state being migrated.
57	* Instead of RDMA_WRITE messages, we use RDMA_SEND
58	* messages for that state, which requires a different
59	* delivery design than main memory.
60	*/
61	#define RDMA_SEND_INCREMENT 32768
62
63	/*
64	* Maximum size infiniband SEND message
65	*/
66	#define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
67	#define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
68
69	#define RDMA_CONTROL_VERSION_CURRENT 1
70	/*
71	* Capabilities for negotiation.
72	*/
73	#define RDMA_CAPABILITY_PIN_ALL 0x01
74
75	/*
76	* Add the other flags above to this list of known capabilities
77	* as they are introduced.
78	*/
79	static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL;
80
81	/*
82	* A work request ID is 64-bits and we split up these bits
83	* into 3 parts:
84	*
85	* bits 0-15 : type of control message, 2^16
86	* bits 16-29: ram block index, 2^14
87	* bits 30-63: ram block chunk number, 2^34
88	*
89	* The last two bit ranges are only used for RDMA writes,
90	* in order to track their completion and potentially
91	* also track unregistration status of the message.
92	*/
93	#define RDMA_WRID_TYPE_SHIFT 0UL
94	#define RDMA_WRID_BLOCK_SHIFT 16UL
95	#define RDMA_WRID_CHUNK_SHIFT 30UL
96
97	#define RDMA_WRID_TYPE_MASK \
98	((1UL << RDMA_WRID_BLOCK_SHIFT) - 1UL)
99
100	#define RDMA_WRID_BLOCK_MASK \
101	(~RDMA_WRID_TYPE_MASK & ((1UL << RDMA_WRID_CHUNK_SHIFT) - 1UL))
102
103	#define RDMA_WRID_CHUNK_MASK (~RDMA_WRID_BLOCK_MASK & ~RDMA_WRID_TYPE_MASK)
104
105	/*
106	* RDMA migration protocol:
107	* 1. RDMA Writes (data messages, i.e. RAM)
108	* 2. IB Send/Recv (control channel messages)
109	*/
110	enum {
111	RDMA_WRID_NONE = 0,
112	RDMA_WRID_RDMA_WRITE = 1,
113	RDMA_WRID_SEND_CONTROL = 2000,
114	RDMA_WRID_RECV_CONTROL = 4000,
115	};
116
117	/*
118	* Work request IDs for IB SEND messages only (not RDMA writes).
119	* This is used by the migration protocol to transmit
120	* control messages (such as device state and registration commands)
121	*
122	* We could use more WRs, but we have enough for now.
123	*/
124	enum {
125	RDMA_WRID_READY = 0,
126	RDMA_WRID_DATA,
127	RDMA_WRID_CONTROL,
128	RDMA_WRID_MAX,
129	};
130
131	/*
132	* SEND/RECV IB Control Messages.
133	*/
134	enum {
135	RDMA_CONTROL_NONE = 0,
136	RDMA_CONTROL_ERROR,
137	RDMA_CONTROL_READY, /* ready to receive */
138	RDMA_CONTROL_QEMU_FILE, /* QEMUFile-transmitted bytes */
139	RDMA_CONTROL_RAM_BLOCKS_REQUEST, /* RAMBlock synchronization */
140	RDMA_CONTROL_RAM_BLOCKS_RESULT, /* RAMBlock synchronization */
141	RDMA_CONTROL_COMPRESS, /* page contains repeat values */
142	RDMA_CONTROL_REGISTER_REQUEST, /* dynamic page registration */
143	RDMA_CONTROL_REGISTER_RESULT, /* key to use after registration */
144	RDMA_CONTROL_REGISTER_FINISHED, /* current iteration finished */
145	RDMA_CONTROL_NUM,
146	};
147
148
149	/*
150	* Memory and MR structures used to represent an IB Send/Recv work request.
151	* This is not used for RDMA writes, only IB Send/Recv.
152	*/
153	typedef struct {
154	uint8_t control[RDMA_CONTROL_MAX_BUFFER]; /* actual buffer to register */
155	struct ibv_mr control_mr; / registration metadata */
156	size_t control_len; /* length of the message */
157	uint8_t control_curr; / start of unconsumed bytes */
158	} RDMAWorkRequestData;
159
160	/*
161	* Negotiate RDMA capabilities during connection-setup time.
162	*/
163	typedef struct {
164	uint32_t version;
165	uint32_t flags;
166	} RDMACapabilities;
167
168	static void caps_to_network(RDMACapabilities *cap)
169	{
170	cap->version = htonl(cap->version);
171	cap->flags = htonl(cap->flags);
172	}
173
174	static void network_to_caps(RDMACapabilities *cap)
175	{
176	cap->version = ntohl(cap->version);
177	cap->flags = ntohl(cap->flags);
178	}
179
180	/*
181	* Representation of a RAMBlock from an RDMA perspective.
182	* This is not transmitted, only local.
183	* This and subsequent structures cannot be linked lists
184	* because we're using a single IB message to transmit
185	* the information. It's small anyway, so a list is overkill.
186	*/
187	typedef struct RDMALocalBlock {
188	char *block_name;
189	uint8_t local_host_addr; / local virtual address */
190	uint64_t remote_host_addr; /* remote virtual address */
191	uint64_t offset;
192	uint64_t length;
193	struct ibv_mr *pmr; / MRs for chunk-level registration */
194	struct ibv_mr mr; / MR for non-chunk-level registration */
195	uint32_t remote_keys; / rkeys for chunk-level registration */
196	uint32_t remote_rkey; /* rkeys for non-chunk-level registration */
197	int index; /* which block are we */
198	unsigned int src_index; /* (Only used on dest) */
199	int nb_chunks;
200	unsigned long *transit_bitmap;
201	} RDMALocalBlock;
202
203	/*
204	* Also represents a RAMblock, but only on the dest.
205	* This gets transmitted by the dest during connection-time
206	* to the source VM and then is used to populate the
207	* corresponding RDMALocalBlock with
208	* the information needed to perform the actual RDMA.
209	*/
210	typedef struct QEMU_PACKED RDMADestBlock {
211	uint64_t remote_host_addr;
212	uint64_t offset;
213	uint64_t length;
214	uint32_t remote_rkey;
215	uint32_t padding;
216	} RDMADestBlock;
217
218	static const char *control_desc(unsigned int rdma_control)
219	{
220	static const char *strs[] = {
221	[RDMA_CONTROL_NONE] = "NONE",
222	[RDMA_CONTROL_ERROR] = "ERROR",
223	[RDMA_CONTROL_READY] = "READY",
224	[RDMA_CONTROL_QEMU_FILE] = "QEMU FILE",
225	[RDMA_CONTROL_RAM_BLOCKS_REQUEST] = "RAM BLOCKS REQUEST",
226	[RDMA_CONTROL_RAM_BLOCKS_RESULT] = "RAM BLOCKS RESULT",
227	[RDMA_CONTROL_COMPRESS] = "COMPRESS",
228	[RDMA_CONTROL_REGISTER_REQUEST] = "REGISTER REQUEST",
229	[RDMA_CONTROL_REGISTER_RESULT] = "REGISTER RESULT",
230	[RDMA_CONTROL_REGISTER_FINISHED] = "REGISTER FINISHED",
231	};
232
233	if (rdma_control >= RDMA_CONTROL_NUM) {
234	return "??BAD CONTROL VALUE??";
235	}
236
237	return strs[rdma_control];
238	}
239
240	#if !defined(CONFIG_ARPA_INET_64)
241	static uint64_t htonll(uint64_t v)
242	{
243	union { uint32_t lv[2]; uint64_t llv; } u;
244	u.lv[0] = htonl(v >> 32);
245	u.lv[1] = htonl(v & 0xFFFFFFFFULL);
246	return u.llv;
247	}
248
249	static uint64_t ntohll(uint64_t v)
250	{
251	union { uint32_t lv[2]; uint64_t llv; } u;
252	u.llv = v;
253	return ((uint64_t)ntohl(u.lv[0]) << 32) \| (uint64_t) ntohl(u.lv[1]);
254	}
255	#endif
256
257	static void dest_block_to_network(RDMADestBlock *db)
258	{
259	db->remote_host_addr = htonll(db->remote_host_addr);
260	db->offset = htonll(db->offset);
261	db->length = htonll(db->length);
262	db->remote_rkey = htonl(db->remote_rkey);
263	}
264
265	static void network_to_dest_block(RDMADestBlock *db)
266	{
267	db->remote_host_addr = ntohll(db->remote_host_addr);
268	db->offset = ntohll(db->offset);
269	db->length = ntohll(db->length);
270	db->remote_rkey = ntohl(db->remote_rkey);
271	}
272
273	/*
274	* Virtual address of the above structures used for transmitting
275	* the RAMBlock descriptions at connection-time.
276	* This structure is not transmitted.
277	*/
278	typedef struct RDMALocalBlocks {
279	int nb_blocks;
280	RDMALocalBlock *block;
281	} RDMALocalBlocks;
282
283	/*
284	* Main data structure for RDMA state.
285	* While there is only one copy of this structure being allocated right now,
286	* this is the place where one would start if you wanted to consider
287	* having more than one RDMA connection open at the same time.
288	*/
289	typedef struct RDMAContext {
290	char *host;
291	int port;
292
293	RDMAWorkRequestData wr_data[RDMA_WRID_MAX];
294
295	/*
296	* This is used by *_exchange_send() to figure out whether or not
297	* the initial "READY" message has already been received or not.
298	* This is because other functions may potentially poll() and detect
299	* the READY message before send() does, in which case we need to
300	* know if it completed.
301	*/
302	int control_ready_expected;
303
304	/* number of outstanding writes */
305	int nb_sent;
306
307	/* store info about current buffer so that we can
308	merge it with future sends */
309	uint64_t current_addr;
310	uint64_t current_length;
311	/* index of ram block the current buffer belongs to */
312	int current_index;
313	/* index of the chunk in the current ram block */
314	int current_chunk;
315
316	bool pin_all;
317
318	/*
319	* infiniband-specific variables for opening the device
320	* and maintaining connection state and so forth.
321	*
322	* cm_id also has ibv_context, rdma_event_channel, and ibv_qp in
323	* cm_id->verbs, cm_id->channel, and cm_id->qp.
324	*/
325	struct rdma_cm_id cm_id; / connection manager ID */
326	struct rdma_cm_id *listen_id;
327	bool connected;
328
329	struct ibv_context *verbs;
330	struct rdma_event_channel *channel;
331	struct ibv_qp qp; / queue pair */
332	struct ibv_comp_channel recv_comp_channel; / recv completion channel */
333	struct ibv_comp_channel send_comp_channel; / send completion channel */
334	struct ibv_pd pd; / protection domain */
335	struct ibv_cq recv_cq; / recvieve completion queue */
336	struct ibv_cq send_cq; / send completion queue */
337
338	/*
339	* If a previous write failed (perhaps because of a failed
340	* memory registration, then do not attempt any future work
341	* and remember the error state.
342	*/
343	bool errored;
344	bool error_reported;
345	bool received_error;
346
347	/*
348	* Description of ram blocks used throughout the code.
349	*/
350	RDMALocalBlocks local_ram_blocks;
351	RDMADestBlock *dest_blocks;
352
353	/* Index of the next RAMBlock received during block registration */
354	unsigned int next_src_index;
355
356	/*
357	* Migration on destination started.
358	* Then use coroutine yield function.
359	* Source runs in a thread, so we don't care.
360	*/
361	int migration_started_on_destination;
362
363	int total_registrations;
364	int total_writes;
365
366	GHashTable *blockmap;
367
368	/* the RDMAContext for return path */
369	struct RDMAContext *return_path;
370	bool is_return_path;
371	} RDMAContext;
372
373	#define TYPE_QIO_CHANNEL_RDMA "qio-channel-rdma"
374	OBJECT_DECLARE_SIMPLE_TYPE(QIOChannelRDMA, QIO_CHANNEL_RDMA)
375
376
377
378	struct QIOChannelRDMA {
379	QIOChannel parent;
380	RDMAContext *rdmain;
381	RDMAContext *rdmaout;
382	bool blocking; /* XXX we don't actually honour this yet */
383	};
384
385	/*
386	* Main structure for IB Send/Recv control messages.
387	* This gets prepended at the beginning of every Send/Recv.
388	*/
389	typedef struct QEMU_PACKED {
390	uint32_t len; /* Total length of data portion */
391	uint32_t type; /* which control command to perform */
392	uint32_t repeat; /* number of commands in data portion of same type */
393	uint32_t padding;
394	} RDMAControlHeader;
395
396	static void control_to_network(RDMAControlHeader *control)
397	{
398	control->type = htonl(control->type);
399	control->len = htonl(control->len);
400	control->repeat = htonl(control->repeat);
401	}
402
403	static void network_to_control(RDMAControlHeader *control)
404	{
405	control->type = ntohl(control->type);
406	control->len = ntohl(control->len);
407	control->repeat = ntohl(control->repeat);
408	}
409
410	/*
411	* Register a single Chunk.
412	* Information sent by the source VM to inform the dest
413	* to register an single chunk of memory before we can perform
414	* the actual RDMA operation.
415	*/
416	typedef struct QEMU_PACKED {
417	uint64_t current_addr; /* offset into the ram_addr_t space */
418	uint32_t current_index; /* which ramblock the chunk belongs to */
419	uint32_t padding;
420	uint64_t chunks; /* how many sequential chunks to register */
421	} RDMARegister;
422
423	static bool rdma_errored(RDMAContext *rdma)
424	{
425	if (rdma->errored && !rdma->error_reported) {
426	error_report("RDMA is in an error state waiting migration"
427	" to abort!");
428	rdma->error_reported = true;
429	}
430	return rdma->errored;
431	}
432
433	static void register_to_network(RDMAContext rdma, RDMARegister reg)
434	{
435	RDMALocalBlock *local_block;
436	local_block = &rdma->local_ram_blocks.block[reg->current_index];
437
438	/*
439	* current_addr as passed in is an address in the local ram_addr_t
440	* space, we need to translate this for the destination
441	*/
442	reg->current_addr -= local_block->offset;
443	reg->current_addr += rdma->dest_blocks[reg->current_index].offset;
444	reg->current_addr = htonll(reg->current_addr);
445	reg->current_index = htonl(reg->current_index);
446	reg->chunks = htonll(reg->chunks);
447	}
448
449	static void network_to_register(RDMARegister *reg)
450	{
451	reg->current_addr = ntohll(reg->current_addr);
452	reg->current_index = ntohl(reg->current_index);
453	reg->chunks = ntohll(reg->chunks);
454	}
455
456	typedef struct QEMU_PACKED {
457	uint32_t value; /* if zero, we will madvise() */
458	uint32_t block_idx; /* which ram block index */
459	uint64_t offset; /* Address in remote ram_addr_t space */
460	uint64_t length; /* length of the chunk */
461	} RDMACompress;
462
463	static void compress_to_network(RDMAContext rdma, RDMACompress comp)
464	{
465	comp->value = htonl(comp->value);
466	/*
467	* comp->offset as passed in is an address in the local ram_addr_t
468	* space, we need to translate this for the destination
469	*/
470	comp->offset -= rdma->local_ram_blocks.block[comp->block_idx].offset;
471	comp->offset += rdma->dest_blocks[comp->block_idx].offset;
472	comp->block_idx = htonl(comp->block_idx);
473	comp->offset = htonll(comp->offset);
474	comp->length = htonll(comp->length);
475	}
476
477	static void network_to_compress(RDMACompress *comp)
478	{
479	comp->value = ntohl(comp->value);
480	comp->block_idx = ntohl(comp->block_idx);
481	comp->offset = ntohll(comp->offset);
482	comp->length = ntohll(comp->length);
483	}
484
485	/*
486	* The result of the dest's memory registration produces an "rkey"
487	* which the source VM must reference in order to perform
488	* the RDMA operation.
489	*/
490	typedef struct QEMU_PACKED {
491	uint32_t rkey;
492	uint32_t padding;
493	uint64_t host_addr;
494	} RDMARegisterResult;
495
496	static void result_to_network(RDMARegisterResult *result)
497	{
498	result->rkey = htonl(result->rkey);
499	result->host_addr = htonll(result->host_addr);
500	};
501
502	static void network_to_result(RDMARegisterResult *result)
503	{
504	result->rkey = ntohl(result->rkey);
505	result->host_addr = ntohll(result->host_addr);
506	};
507
508	static int qemu_rdma_exchange_send(RDMAContext rdma, RDMAControlHeader head,
509	uint8_t data, RDMAControlHeader resp,
510	int *resp_idx,
511	int (callback)(RDMAContext rdma,
512	Error **errp),
513	Error **errp);
514
515	static inline uint64_t ram_chunk_index(const uint8_t *start,
516	const uint8_t *host)
517	{
518	return ((uintptr_t) host - (uintptr_t) start) / migrate_rdma_chunk_size();
519	}
520
521	static inline uint8_t ram_chunk_start(const RDMALocalBlock rdma_ram_block,
522	uint64_t i)
523	{
524	return (uint8_t *)(uintptr_t)(rdma_ram_block->local_host_addr +
525	(i * migrate_rdma_chunk_size()));
526	}
527
528	static inline uint8_t ram_chunk_end(const RDMALocalBlock rdma_ram_block,
529	uint64_t i)
530	{
531	uint8_t *result = ram_chunk_start(rdma_ram_block, i) +
532	migrate_rdma_chunk_size();
533
534	if (result > (rdma_ram_block->local_host_addr + rdma_ram_block->length)) {
535	result = rdma_ram_block->local_host_addr + rdma_ram_block->length;
536	}
537
538	return result;
539	}
540
541	static void rdma_add_block(RDMAContext rdma, const char block_name,
542	void *host_addr,
543	ram_addr_t block_offset, uint64_t length)
544	{
545	RDMALocalBlocks *local = &rdma->local_ram_blocks;
546	RDMALocalBlock *block;
547	RDMALocalBlock *old = local->block;
548
549	local->block = g_new0(RDMALocalBlock, local->nb_blocks + 1);
550
551	if (local->nb_blocks) {
552	if (rdma->blockmap) {
553	for (int x = 0; x < local->nb_blocks; x++) {
554	g_hash_table_remove(rdma->blockmap,
555	(void *)(uintptr_t)old[x].offset);
556	g_hash_table_insert(rdma->blockmap,
557	(void *)(uintptr_t)old[x].offset,
558	&local->block[x]);
559	}
560	}
561	memcpy(local->block, old, sizeof(RDMALocalBlock) * local->nb_blocks);
562	g_free(old);
563	}
564
565	block = &local->block[local->nb_blocks];
566
567	block->block_name = g_strdup(block_name);
568	block->local_host_addr = host_addr;
569	block->offset = block_offset;
570	block->length = length;
571	block->index = local->nb_blocks;
572	block->src_index = ~0U; /* Filled in by the receipt of the block list */
573	block->nb_chunks = ram_chunk_index(host_addr, host_addr + length) + 1UL;
574	block->transit_bitmap = bitmap_new(block->nb_chunks);
575	bitmap_clear(block->transit_bitmap, 0, block->nb_chunks);
576	block->remote_keys = g_new0(uint32_t, block->nb_chunks);
577
578	if (rdma->blockmap) {
579	g_hash_table_insert(rdma->blockmap, (void *)(uintptr_t)block_offset, block);
580	}
581
582	trace_rdma_add_block(block_name, local->nb_blocks,
583	(uintptr_t) block->local_host_addr,
584	block->offset, block->length,
585	(uintptr_t) (block->local_host_addr + block->length),
586	BITS_TO_LONGS(block->nb_chunks) *
587	sizeof(unsigned long) * 8,
588	block->nb_chunks);
589
590	local->nb_blocks++;
591	}
592
593	/*
594	* Memory regions need to be registered with the device and queue pairs setup
595	* in advanced before the migration starts. This tells us where the RAM blocks
596	* are so that we can register them individually.
597	*/
598	static int qemu_rdma_init_one_block(RAMBlock rb, void opaque)
599	{
600	const char *block_name = qemu_ram_get_idstr(rb);
601	void *host_addr = qemu_ram_get_host_addr(rb);
602	ram_addr_t block_offset = qemu_ram_get_offset(rb);
603	ram_addr_t length = qemu_ram_get_used_length(rb);
604	rdma_add_block(opaque, block_name, host_addr, block_offset, length);
605	return 0;
606	}
607
608	/*
609	* Identify the RAMBlocks and their quantity. They will be references to
610	* identify chunk boundaries inside each RAMBlock and also be referenced
611	* during dynamic page registration.
612	*/
613	static void qemu_rdma_init_ram_blocks(RDMAContext *rdma)
614	{
615	RDMALocalBlocks *local = &rdma->local_ram_blocks;
616	int ret;
617
618	assert(rdma->blockmap == NULL);
619	memset(local, 0, sizeof *local);
620	ret = foreach_not_ignored_block(qemu_rdma_init_one_block, rdma);
621	assert(!ret);
622	trace_rdma_init_ram_blocks(local->nb_blocks);
623	rdma->dest_blocks = g_new0(RDMADestBlock,
624	rdma->local_ram_blocks.nb_blocks);
625	}
626
627	/*
628	* Note: If used outside of cleanup, the caller must ensure that the destination
629	* block structures are also updated
630	*/
631	static void rdma_delete_block(RDMAContext rdma, RDMALocalBlock block)
632	{
633	RDMALocalBlocks *local = &rdma->local_ram_blocks;
634	RDMALocalBlock *old = local->block;
635
636	if (rdma->blockmap) {
637	g_hash_table_remove(rdma->blockmap, (void *)(uintptr_t)block->offset);
638	}
639	if (block->pmr) {
640	for (int j = 0; j < block->nb_chunks; j++) {
641	if (!block->pmr[j]) {
642	continue;
643	}
644	ibv_dereg_mr(block->pmr[j]);
645	rdma->total_registrations--;
646	}
647	g_free(block->pmr);
648	block->pmr = NULL;
649	}
650
651	if (block->mr) {
652	ibv_dereg_mr(block->mr);
653	rdma->total_registrations--;
654	block->mr = NULL;
655	}
656
657	g_free(block->transit_bitmap);
658	block->transit_bitmap = NULL;
659
660	g_free(block->remote_keys);
661	block->remote_keys = NULL;
662
663	g_free(block->block_name);
664	block->block_name = NULL;
665
666	if (rdma->blockmap) {
667	for (int x = 0; x < local->nb_blocks; x++) {
668	g_hash_table_remove(rdma->blockmap,
669	(void *)(uintptr_t)old[x].offset);
670	}
671	}
672
673	if (local->nb_blocks > 1) {
674
675	local->block = g_new0(RDMALocalBlock, local->nb_blocks - 1);
676
677	if (block->index) {
678	memcpy(local->block, old, sizeof(RDMALocalBlock) * block->index);
679	}
680
681	if (block->index < (local->nb_blocks - 1)) {
682	memcpy(local->block + block->index, old + (block->index + 1),
683	sizeof(RDMALocalBlock) *
684	(local->nb_blocks - (block->index + 1)));
685	for (int x = block->index; x < local->nb_blocks - 1; x++) {
686	local->block[x].index--;
687	}
688	}
689	} else {
690	assert(block == local->block);
691	local->block = NULL;
692	}
693
694	trace_rdma_delete_block(block, (uintptr_t)block->local_host_addr,
695	block->offset, block->length,
696	(uintptr_t)(block->local_host_addr + block->length),
697	BITS_TO_LONGS(block->nb_chunks) *
698	sizeof(unsigned long) * 8, block->nb_chunks);
699
700	g_free(old);
701
702	local->nb_blocks--;
703
704	if (local->nb_blocks && rdma->blockmap) {
705	for (int x = 0; x < local->nb_blocks; x++) {
706	g_hash_table_insert(rdma->blockmap,
707	(void *)(uintptr_t)local->block[x].offset,
708	&local->block[x]);
709	}
710	}
711	}
712
713	/*
714	* Trace RDMA device open, with device details.
715	*/
716	static void qemu_rdma_dump_id(const char who, struct ibv_context verbs)
717	{
718	struct ibv_port_attr port;
719
720	if (ibv_query_port(verbs, 1, &port)) {
721	trace_rdma_dump_id_failed(who);
722	return;
723	}
724
725	trace_rdma_dump_id(who,
726	verbs->device->name,
727	verbs->device->dev_name,
728	verbs->device->dev_path,
729	verbs->device->ibdev_path,
730	port.link_layer,
731	port.link_layer == IBV_LINK_LAYER_INFINIBAND ? "Infiniband"
732	: port.link_layer == IBV_LINK_LAYER_ETHERNET ? "Ethernet"
733	: "Unknown");
734	}
735
736	/*
737	* Trace RDMA gid addressing information.
738	* Useful for understanding the RDMA device hierarchy in the kernel.
739	*/
740	static void qemu_rdma_dump_gid(const char who, struct rdma_cm_id id)
741	{
742	char sgid[33];
743	char dgid[33];
744	inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.sgid, sgid, sizeof sgid);
745	inet_ntop(AF_INET6, &id->route.addr.addr.ibaddr.dgid, dgid, sizeof dgid);
746	trace_rdma_dump_gid(who, sgid, dgid);
747	}
748
749	/*
750	* Figure out which RDMA device corresponds to the requested IP hostname
751	* Also create the initial connection manager identifiers for opening
752	* the connection.
753	*/
754	static int qemu_rdma_resolve_host(RDMAContext rdma, Error *errp)
755	{
756	int ret;
757	struct rdma_addrinfo *res;
758	char port_str[16];
759	struct rdma_cm_event *cm_event;
760	char ip[40] = "unknown";
761
762	if (rdma->host == NULL \|\| !strcmp(rdma->host, "")) {
763	error_setg(errp, "RDMA ERROR: RDMA hostname has not been set");
764	return -1;
765	}
766
767	/* create CM channel */
768	rdma->channel = rdma_create_event_channel();
769	if (!rdma->channel) {
770	error_setg(errp, "RDMA ERROR: could not create CM channel");
771	return -1;
772	}
773
774	/* create CM id */
775	ret = rdma_create_id(rdma->channel, &rdma->cm_id, NULL, RDMA_PS_TCP);
776	if (ret < 0) {
777	error_setg(errp, "RDMA ERROR: could not create channel id");
778	goto err_resolve_create_id;
779	}
780
781	snprintf(port_str, 16, "%d", rdma->port);
782	port_str[15] = '\0';
783
784	ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
785	if (ret) {
786	error_setg(errp, "RDMA ERROR: could not rdma_getaddrinfo address %s",
787	rdma->host);
788	goto err_resolve_get_addr;
789	}
790
791	/* Try all addresses, exit loop on first success of resolving address */
792	for (struct rdma_addrinfo *e = res; e != NULL; e = e->ai_next) {
793
794	inet_ntop(e->ai_family,
795	&((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
796	trace_rdma_resolve_host_trying(rdma->host, ip);
797
798	ret = rdma_resolve_addr(rdma->cm_id, NULL, e->ai_dst_addr,
799	RDMA_RESOLVE_TIMEOUT_MS);
800	if (ret >= 0) {
801	goto route;
802	}
803	}
804
805	rdma_freeaddrinfo(res);
806	error_setg(errp, "RDMA ERROR: could not resolve address %s", rdma->host);
807	goto err_resolve_get_addr;
808
809	route:
810	rdma_freeaddrinfo(res);
811	qemu_rdma_dump_gid("source_resolve_addr", rdma->cm_id);
812
813	ret = rdma_get_cm_event(rdma->channel, &cm_event);
814	if (ret < 0) {
815	error_setg(errp, "RDMA ERROR: could not perform event_addr_resolved");
816	goto err_resolve_get_addr;
817	}
818
819	if (cm_event->event != RDMA_CM_EVENT_ADDR_RESOLVED) {
820	error_setg(errp,
821	"RDMA ERROR: result not equal to event_addr_resolved %s",
822	rdma_event_str(cm_event->event));
823	rdma_ack_cm_event(cm_event);
824	goto err_resolve_get_addr;
825	}
826	rdma_ack_cm_event(cm_event);
827
828	/* resolve route */
829	ret = rdma_resolve_route(rdma->cm_id, RDMA_RESOLVE_TIMEOUT_MS);
830	if (ret < 0) {
831	error_setg(errp, "RDMA ERROR: could not resolve rdma route");
832	goto err_resolve_get_addr;
833	}
834
835	ret = rdma_get_cm_event(rdma->channel, &cm_event);
836	if (ret < 0) {
837	error_setg(errp, "RDMA ERROR: could not perform event_route_resolved");
838	goto err_resolve_get_addr;
839	}
840	if (cm_event->event != RDMA_CM_EVENT_ROUTE_RESOLVED) {
841	error_setg(errp, "RDMA ERROR: "
842	"result not equal to event_route_resolved: %s",
843	rdma_event_str(cm_event->event));
844	rdma_ack_cm_event(cm_event);
845	goto err_resolve_get_addr;
846	}
847	rdma_ack_cm_event(cm_event);
848	rdma->verbs = rdma->cm_id->verbs;
849	qemu_rdma_dump_id("source_resolve_host", rdma->cm_id->verbs);
850	qemu_rdma_dump_gid("source_resolve_host", rdma->cm_id);
851	return 0;
852
853	err_resolve_get_addr:
854	rdma_destroy_id(rdma->cm_id);
855	rdma->cm_id = NULL;
856	err_resolve_create_id:
857	rdma_destroy_event_channel(rdma->channel);
858	rdma->channel = NULL;
859	return -1;
860	}
861
862	/*
863	* Create protection domain and completion queues
864	*/
865	static int qemu_rdma_alloc_pd_cq(RDMAContext rdma, Error *errp)
866	{
867	/* allocate pd */
868	rdma->pd = ibv_alloc_pd(rdma->verbs);
869	if (!rdma->pd) {
870	error_setg(errp, "failed to allocate protection domain");
871	return -1;
872	}
873
874	/* create receive completion channel */
875	rdma->recv_comp_channel = ibv_create_comp_channel(rdma->verbs);
876	if (!rdma->recv_comp_channel) {
877	error_setg(errp, "failed to allocate receive completion channel");
878	goto err_alloc_pd_cq;
879	}
880
881	/*
882	* Completion queue can be filled by read work requests.
883	*/
884	rdma->recv_cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
885	NULL, rdma->recv_comp_channel, 0);
886	if (!rdma->recv_cq) {
887	error_setg(errp, "failed to allocate receive completion queue");
888	goto err_alloc_pd_cq;
889	}
890
891	/* create send completion channel */
892	rdma->send_comp_channel = ibv_create_comp_channel(rdma->verbs);
893	if (!rdma->send_comp_channel) {
894	error_setg(errp, "failed to allocate send completion channel");
895	goto err_alloc_pd_cq;
896	}
897
898	rdma->send_cq = ibv_create_cq(rdma->verbs, (RDMA_SIGNALED_SEND_MAX * 3),
899	NULL, rdma->send_comp_channel, 0);
900	if (!rdma->send_cq) {
901	error_setg(errp, "failed to allocate send completion queue");
902	goto err_alloc_pd_cq;
903	}
904
905	return 0;
906
907	err_alloc_pd_cq:
908	if (rdma->pd) {
909	ibv_dealloc_pd(rdma->pd);
910	}
911	if (rdma->recv_comp_channel) {
912	ibv_destroy_comp_channel(rdma->recv_comp_channel);
913	}
914	if (rdma->send_comp_channel) {
915	ibv_destroy_comp_channel(rdma->send_comp_channel);
916	}
917	if (rdma->recv_cq) {
918	ibv_destroy_cq(rdma->recv_cq);
919	rdma->recv_cq = NULL;
920	}
921	rdma->pd = NULL;
922	rdma->recv_comp_channel = NULL;
923	rdma->send_comp_channel = NULL;
924	return -1;
925
926	}
927
928	/*
929	* Create queue pairs.
930	*/
931	static int qemu_rdma_alloc_qp(RDMAContext *rdma)
932	{
933	struct ibv_qp_init_attr attr = { 0 };
934
935	attr.cap.max_send_wr = RDMA_SIGNALED_SEND_MAX;
936	attr.cap.max_recv_wr = 3;
937	attr.cap.max_send_sge = 1;
938	attr.cap.max_recv_sge = 1;
939	attr.send_cq = rdma->send_cq;
940	attr.recv_cq = rdma->recv_cq;
941	attr.qp_type = IBV_QPT_RC;
942
943	if (rdma_create_qp(rdma->cm_id, rdma->pd, &attr) < 0) {
944	return -1;
945	}
946
947	rdma->qp = rdma->cm_id->qp;
948	return 0;
949	}
950
951	/* Check whether On-Demand Paging is supported by RDAM device */
952	static bool rdma_support_odp(struct ibv_context *dev)
953	{
954	struct ibv_device_attr_ex attr = {0};
955
956	if (ibv_query_device_ex(dev, NULL, &attr)) {
957	return false;
958	}
959
960	if (attr.odp_caps.general_caps & IBV_ODP_SUPPORT) {
961	return true;
962	}
963
964	return false;
965	}
966
967	/*
968	* ibv_advise_mr to avoid RNR NAK error as far as possible.
969	* The responder mr registering with ODP will sent RNR NAK back to
970	* the requester in the face of the page fault.
971	*/
972	static void qemu_rdma_advise_prefetch_mr(struct ibv_pd *pd, uint64_t addr,
973	uint32_t len, uint32_t lkey,
974	const char *name, bool wr)
975	{
976	#ifdef HAVE_IBV_ADVISE_MR
977	int ret;
978	int advice = wr ? IBV_ADVISE_MR_ADVICE_PREFETCH_WRITE :
979	IBV_ADVISE_MR_ADVICE_PREFETCH;
980	struct ibv_sge sg_list = {.lkey = lkey, .addr = addr, .length = len};
981
982	ret = ibv_advise_mr(pd, advice,
983	IBV_ADVISE_MR_FLAG_FLUSH, &sg_list, 1);
984	/* ignore the error */
985	trace_rdma_advise_mr(name, len, addr, strerror(ret));
986	#endif
987	}
988
989	static int qemu_rdma_reg_whole_ram_blocks(RDMAContext rdma, Error *errp)
990	{
991	int i;
992	RDMALocalBlocks *local = &rdma->local_ram_blocks;
993
994	for (i = 0; i < local->nb_blocks; i++) {
995	int access = IBV_ACCESS_LOCAL_WRITE \| IBV_ACCESS_REMOTE_WRITE;
996
997	local->block[i].mr =
998	ibv_reg_mr(rdma->pd,
999	local->block[i].local_host_addr,
1000	local->block[i].length, access
1001	);
1002	/*
1003	* ibv_reg_mr() is not documented to set errno. If it does,
1004	* it's somebody else's doc bug. If it doesn't, the use of
1005	* errno below is wrong.
1006	* TODO Find out whether ibv_reg_mr() sets errno.
1007	*/
1008	if (!local->block[i].mr &&
1009	errno == ENOTSUP && rdma_support_odp(rdma->verbs)) {
1010	access \|= IBV_ACCESS_ON_DEMAND;
1011	/* register ODP mr */
1012	local->block[i].mr =
1013	ibv_reg_mr(rdma->pd,
1014	local->block[i].local_host_addr,
1015	local->block[i].length, access);
1016	trace_rdma_register_odp_mr(local->block[i].block_name);
1017
1018	if (local->block[i].mr) {
1019	qemu_rdma_advise_prefetch_mr(rdma->pd,
1020	(uintptr_t)local->block[i].local_host_addr,
1021	local->block[i].length,
1022	local->block[i].mr->lkey,
1023	local->block[i].block_name,
1024	true);
1025	}
1026	}
1027
1028	if (!local->block[i].mr) {
1029	error_setg_errno(errp, errno,
1030	"Failed to register local dest ram block!");
1031	goto err;
1032	}
1033	rdma->total_registrations++;
1034	}
1035
1036	return 0;
1037
1038	err:
1039	for (i--; i >= 0; i--) {
1040	ibv_dereg_mr(local->block[i].mr);
1041	local->block[i].mr = NULL;
1042	rdma->total_registrations--;
1043	}
1044
1045	return -1;
1046
1047	}
1048
1049	/*
1050	* Find the ram block that corresponds to the page requested to be
1051	* transmitted by QEMU.
1052	*
1053	* Once the block is found, also identify which 'chunk' within that
1054	* block that the page belongs to.
1055	*/
1056	static void qemu_rdma_search_ram_block(RDMAContext *rdma,
1057	uintptr_t block_offset,
1058	uint64_t offset,
1059	uint64_t length,
1060	uint64_t *block_index,
1061	uint64_t *chunk_index)
1062	{
1063	uint64_t current_addr = block_offset + offset;
1064	RDMALocalBlock *block = g_hash_table_lookup(rdma->blockmap,
1065	(void *) block_offset);
1066	assert(block);
1067	assert(current_addr >= block->offset);
1068	assert((current_addr + length) <= (block->offset + block->length));
1069
1070	*block_index = block->index;
1071	*chunk_index = ram_chunk_index(block->local_host_addr,
1072	block->local_host_addr + (current_addr - block->offset));
1073	}
1074
1075	/*
1076	* Register a chunk with IB. If the chunk was already registered
1077	* previously, then skip.
1078	*
1079	* Also return the keys associated with the registration needed
1080	* to perform the actual RDMA operation.
1081	*/
1082	static int qemu_rdma_register_and_get_keys(RDMAContext *rdma,
1083	RDMALocalBlock *block, uintptr_t host_addr,
1084	uint32_t lkey, uint32_t rkey, int chunk,
1085	uint8_t chunk_start, uint8_t chunk_end)
1086	{
1087	if (block->mr) {
1088	if (lkey) {
1089	*lkey = block->mr->lkey;
1090	}
1091	if (rkey) {
1092	*rkey = block->mr->rkey;
1093	}
1094	return 0;
1095	}
1096
1097	/* allocate memory to store chunk MRs */
1098	if (!block->pmr) {
1099	block->pmr = g_new0(struct ibv_mr *, block->nb_chunks);
1100	}
1101
1102	/*
1103	* If 'rkey', then we're the destination, so grant access to the source.
1104	*
1105	* If 'lkey', then we're the source VM, so grant access only to ourselves.
1106	*/
1107	if (!block->pmr[chunk]) {
1108	uint64_t len = chunk_end - chunk_start;
1109	int access = rkey ? IBV_ACCESS_LOCAL_WRITE \| IBV_ACCESS_REMOTE_WRITE :
1110	0;
1111
1112	trace_rdma_register_and_get_keys(len, chunk_start);
1113
1114	block->pmr[chunk] = ibv_reg_mr(rdma->pd, chunk_start, len, access);
1115	/*
1116	* ibv_reg_mr() is not documented to set errno. If it does,
1117	* it's somebody else's doc bug. If it doesn't, the use of
1118	* errno below is wrong.
1119	* TODO Find out whether ibv_reg_mr() sets errno.
1120	*/
1121	if (!block->pmr[chunk] &&
1122	errno == ENOTSUP && rdma_support_odp(rdma->verbs)) {
1123	access \|= IBV_ACCESS_ON_DEMAND;
1124	/* register ODP mr */
1125	block->pmr[chunk] = ibv_reg_mr(rdma->pd, chunk_start, len, access);
1126	trace_rdma_register_odp_mr(block->block_name);
1127
1128	if (block->pmr[chunk]) {
1129	qemu_rdma_advise_prefetch_mr(rdma->pd, (uintptr_t)chunk_start,
1130	len, block->pmr[chunk]->lkey,
1131	block->block_name, rkey);
1132
1133	}
1134	}
1135	}
1136	if (!block->pmr[chunk]) {
1137	return -1;
1138	}
1139	rdma->total_registrations++;
1140
1141	if (lkey) {
1142	*lkey = block->pmr[chunk]->lkey;
1143	}
1144	if (rkey) {
1145	*rkey = block->pmr[chunk]->rkey;
1146	}
1147	return 0;
1148	}
1149
1150	/*
1151	* Register (at connection time) the memory used for control
1152	* channel messages.
1153	*/
1154	static int qemu_rdma_reg_control(RDMAContext *rdma, int idx)
1155	{
1156	rdma->wr_data[idx].control_mr = ibv_reg_mr(rdma->pd,
1157	rdma->wr_data[idx].control, RDMA_CONTROL_MAX_BUFFER,
1158	IBV_ACCESS_LOCAL_WRITE \| IBV_ACCESS_REMOTE_WRITE);
1159	if (rdma->wr_data[idx].control_mr) {
1160	rdma->total_registrations++;
1161	return 0;
1162	}
1163	return -1;
1164	}
1165
1166	static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
1167	uint64_t chunk)
1168	{
1169	uint64_t result = wr_id & RDMA_WRID_TYPE_MASK;
1170
1171	result \|= (index << RDMA_WRID_BLOCK_SHIFT);
1172	result \|= (chunk << RDMA_WRID_CHUNK_SHIFT);
1173
1174	return result;
1175	}
1176
1177	/*
1178	* Consult the connection manager to see a work request
1179	* (of any kind) has completed.
1180	* Return the work request ID that completed.
1181	*/
1182	static int qemu_rdma_poll(RDMAContext rdma, struct ibv_cq cq,
1183	uint64_t wr_id_out, uint32_t byte_len)
1184	{
1185	int ret;
1186	struct ibv_wc wc;
1187	uint64_t wr_id;
1188
1189	ret = ibv_poll_cq(cq, 1, &wc);
1190
1191	if (!ret) {
1192	*wr_id_out = RDMA_WRID_NONE;
1193	return 0;
1194	}
1195
1196	if (ret < 0) {
1197	return -1;
1198	}
1199
1200	wr_id = wc.wr_id & RDMA_WRID_TYPE_MASK;
1201
1202	if (wc.status != IBV_WC_SUCCESS) {
1203	return -1;
1204	}
1205
1206	if (rdma->control_ready_expected &&
1207	(wr_id >= RDMA_WRID_RECV_CONTROL)) {
1208	trace_rdma_poll_recv(wr_id - RDMA_WRID_RECV_CONTROL, wr_id,
1209	rdma->nb_sent);
1210	rdma->control_ready_expected = 0;
1211	}
1212
1213	if (wr_id == RDMA_WRID_RDMA_WRITE) {
1214	uint64_t chunk =
1215	(wc.wr_id & RDMA_WRID_CHUNK_MASK) >> RDMA_WRID_CHUNK_SHIFT;
1216	uint64_t index =
1217	(wc.wr_id & RDMA_WRID_BLOCK_MASK) >> RDMA_WRID_BLOCK_SHIFT;
1218	RDMALocalBlock *block = &(rdma->local_ram_blocks.block[index]);
1219
1220	trace_rdma_poll_write(wr_id, rdma->nb_sent,
1221	index, chunk, block->local_host_addr,
1222	(void *)(uintptr_t)block->remote_host_addr);
1223
1224	clear_bit(chunk, block->transit_bitmap);
1225
1226	if (rdma->nb_sent > 0) {
1227	rdma->nb_sent--;
1228	}
1229	} else {
1230	trace_rdma_poll_other(wr_id, rdma->nb_sent);
1231	}
1232
1233	*wr_id_out = wc.wr_id;
1234	if (byte_len) {
1235	*byte_len = wc.byte_len;
1236	}
1237
1238	return 0;
1239	}
1240
1241	/* Wait for activity on the completion channel.
1242	* Returns 0 on success, none-0 on error.
1243	*/
1244	static int coroutine_mixed_fn
1245	qemu_rdma_wait_comp_channel(RDMAContext *rdma,
1246	struct ibv_comp_channel *comp_channel)
1247	{
1248	struct rdma_cm_event *cm_event;
1249
1250	if (qemu_in_coroutine()) {
1251	yield_until_fd_readable(comp_channel->fd);
1252	} else {
1253	/* This is the source side, we're in a separate thread
1254	* or destination prior to migration_fd_process_incoming()
1255	* after postcopy, the destination also in a separate thread.
1256	* we can't yield; so we have to poll the fd.
1257	* But we need to be able to handle 'cancel' or an error
1258	* without hanging forever.
1259	*/
1260	while (!rdma->errored && !rdma->received_error) {
1261	GPollFD pfds[2];
1262	pfds[0].fd = comp_channel->fd;
1263	pfds[0].events = G_IO_IN \| G_IO_HUP \| G_IO_ERR;
1264	pfds[0].revents = 0;
1265
1266	pfds[1].fd = rdma->channel->fd;
1267	pfds[1].events = G_IO_IN \| G_IO_HUP \| G_IO_ERR;
1268	pfds[1].revents = 0;
1269
1270	/* 0.1s timeout, should be fine for a 'cancel' */
1271	switch (qemu_poll_ns(pfds, 2, 100 * 1000 * 1000)) {
1272	case 2:
1273	case 1: /* fd active */
1274	if (pfds[0].revents) {
1275	return 0;
1276	}
1277
1278	if (pfds[1].revents) {
1279	if (rdma_get_cm_event(rdma->channel, &cm_event) < 0) {
1280	return -1;
1281	}
1282
1283	if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED \|\|
1284	cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
1285	rdma_ack_cm_event(cm_event);
1286	return -1;
1287	}
1288	rdma_ack_cm_event(cm_event);
1289	}
1290	break;
1291
1292	case 0: /* Timeout, go around again */
1293	break;
1294
1295	default: /* Error of some type -
1296	* I don't trust errno from qemu_poll_ns
1297	*/
1298	return -1;
1299	}
1300
1301	if (migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) {
1302	/* Bail out and let the cancellation happen */
1303	return -1;
1304	}
1305	}
1306	}
1307
1308	if (rdma->received_error) {
1309	return -1;
1310	}
1311	return -rdma->errored;
1312	}
1313
1314	static struct ibv_comp_channel to_channel(RDMAContext rdma, uint64_t wrid)
1315	{
1316	return wrid < RDMA_WRID_RECV_CONTROL ? rdma->send_comp_channel :
1317	rdma->recv_comp_channel;
1318	}
1319
1320	static struct ibv_cq to_cq(RDMAContext rdma, uint64_t wrid)
1321	{
1322	return wrid < RDMA_WRID_RECV_CONTROL ? rdma->send_cq : rdma->recv_cq;
1323	}
1324
1325	/*
1326	* Block until the next work request has completed.
1327	*
1328	* First poll to see if a work request has already completed,
1329	* otherwise block.
1330	*
1331	* If we encounter completed work requests for IDs other than
1332	* the one we're interested in, then that's generally an error.
1333	*
1334	* The only exception is actual RDMA Write completions. These
1335	* completions only need to be recorded, but do not actually
1336	* need further processing.
1337	*/
1338	static int qemu_rdma_block_for_wrid(RDMAContext *rdma,
1339	uint64_t wrid_requested,
1340	uint32_t *byte_len)
1341	{
1342	int num_cq_events = 0, ret;
1343	struct ibv_cq *cq;
1344	void *cq_ctx;
1345	uint64_t wr_id = RDMA_WRID_NONE, wr_id_in;
1346	struct ibv_comp_channel *ch = to_channel(rdma, wrid_requested);
1347	struct ibv_cq *poll_cq = to_cq(rdma, wrid_requested);
1348
1349	if (ibv_req_notify_cq(poll_cq, 0)) {
1350	return -1;
1351	}
1352	/* poll cq first */
1353	while (wr_id != wrid_requested) {
1354	ret = qemu_rdma_poll(rdma, poll_cq, &wr_id_in, byte_len);
1355	if (ret < 0) {
1356	return -1;
1357	}
1358
1359	wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1360
1361	if (wr_id == RDMA_WRID_NONE) {
1362	break;
1363	}
1364	if (wr_id != wrid_requested) {
1365	trace_rdma_block_for_wrid_miss(wrid_requested, wr_id);
1366	}
1367	}
1368
1369	if (wr_id == wrid_requested) {
1370	return 0;
1371	}
1372
1373	while (1) {
1374	ret = qemu_rdma_wait_comp_channel(rdma, ch);
1375	if (ret < 0) {
1376	goto err_block_for_wrid;
1377	}
1378
1379	ret = ibv_get_cq_event(ch, &cq, &cq_ctx);
1380	if (ret < 0) {
1381	goto err_block_for_wrid;
1382	}
1383
1384	num_cq_events++;
1385
1386	if (ibv_req_notify_cq(cq, 0)) {
1387	goto err_block_for_wrid;
1388	}
1389
1390	while (wr_id != wrid_requested) {
1391	ret = qemu_rdma_poll(rdma, poll_cq, &wr_id_in, byte_len);
1392	if (ret < 0) {
1393	goto err_block_for_wrid;
1394	}
1395
1396	wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
1397
1398	if (wr_id == RDMA_WRID_NONE) {
1399	break;
1400	}
1401	if (wr_id != wrid_requested) {
1402	trace_rdma_block_for_wrid_miss(wrid_requested, wr_id);
1403	}
1404	}
1405
1406	if (wr_id == wrid_requested) {
1407	goto success_block_for_wrid;
1408	}
1409	}
1410
1411	success_block_for_wrid:
1412	if (num_cq_events) {
1413	ibv_ack_cq_events(cq, num_cq_events);
1414	}
1415	return 0;
1416
1417	err_block_for_wrid:
1418	if (num_cq_events) {
1419	ibv_ack_cq_events(cq, num_cq_events);
1420	}
1421
1422	rdma->errored = true;
1423	return -1;
1424	}
1425
1426	/*
1427	* Post a SEND message work request for the control channel
1428	* containing some data and block until the post completes.
1429	*/
1430	static int qemu_rdma_post_send_control(RDMAContext rdma, uint8_t buf,
1431	RDMAControlHeader *head,
1432	Error **errp)
1433	{
1434	int ret;
1435	RDMAWorkRequestData *wr = &rdma->wr_data[RDMA_WRID_CONTROL];
1436	struct ibv_send_wr *bad_wr;
1437	struct ibv_sge sge = {
1438	.addr = (uintptr_t)(wr->control),
1439	.length = head->len + sizeof(RDMAControlHeader),
1440	.lkey = wr->control_mr->lkey,
1441	};
1442	struct ibv_send_wr send_wr = {
1443	.wr_id = RDMA_WRID_SEND_CONTROL,
1444	.opcode = IBV_WR_SEND,
1445	.send_flags = IBV_SEND_SIGNALED,
1446	.sg_list = &sge,
1447	.num_sge = 1,
1448	};
1449
1450	trace_rdma_post_send_control(control_desc(head->type));
1451
1452	/*
1453	* We don't actually need to do a memcpy() in here if we used
1454	* the "sge" properly, but since we're only sending control messages
1455	* (not RAM in a performance-critical path), then its OK for now.
1456	*
1457	* The copy makes the RDMAControlHeader simpler to manipulate
1458	* for the time being.
1459	*/
1460	assert(head->len <= RDMA_CONTROL_MAX_BUFFER - sizeof(*head));
1461	memcpy(wr->control, head, sizeof(RDMAControlHeader));
1462	control_to_network((void *) wr->control);
1463
1464	if (buf) {
1465	memcpy(wr->control + sizeof(RDMAControlHeader), buf, head->len);
1466	}
1467
1468
1469	ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
1470
1471	if (ret > 0) {
1472	error_setg(errp, "Failed to use post IB SEND for control");
1473	return -1;
1474	}
1475
1476	ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_SEND_CONTROL, NULL);
1477	if (ret < 0) {
1478	error_setg(errp, "rdma migration: send polling control error");
1479	return -1;
1480	}
1481
1482	return 0;
1483	}
1484
1485	/*
1486	* Post a RECV work request in anticipation of some future receipt
1487	* of data on the control channel.
1488	*/
1489	static int qemu_rdma_post_recv_control(RDMAContext *rdma, int idx,
1490	Error **errp)
1491	{
1492	struct ibv_recv_wr *bad_wr;
1493	struct ibv_sge sge = {
1494	.addr = (uintptr_t)(rdma->wr_data[idx].control),
1495	.length = RDMA_CONTROL_MAX_BUFFER,
1496	.lkey = rdma->wr_data[idx].control_mr->lkey,
1497	};
1498
1499	struct ibv_recv_wr recv_wr = {
1500	.wr_id = RDMA_WRID_RECV_CONTROL + idx,
1501	.sg_list = &sge,
1502	.num_sge = 1,
1503	};
1504
1505
1506	if (ibv_post_recv(rdma->qp, &recv_wr, &bad_wr)) {
1507	error_setg(errp, "error posting control recv");
1508	return -1;
1509	}
1510
1511	return 0;
1512	}
1513
1514	/*
1515	* Block and wait for a RECV control channel message to arrive.
1516	*/
1517	static int qemu_rdma_exchange_get_response(RDMAContext *rdma,
1518	RDMAControlHeader *head, uint32_t expecting, int idx,
1519	Error **errp)
1520	{
1521	uint32_t byte_len;
1522	int ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RECV_CONTROL + idx,
1523	&byte_len);
1524
1525	if (ret < 0) {
1526	error_setg(errp, "rdma migration: recv polling control error!");
1527	return -1;
1528	}
1529
1530	network_to_control((void *) rdma->wr_data[idx].control);
1531	memcpy(head, rdma->wr_data[idx].control, sizeof(RDMAControlHeader));
1532
1533	trace_rdma_exchange_get_response_start(control_desc(expecting));
1534
1535	if (expecting == RDMA_CONTROL_NONE) {
1536	trace_rdma_exchange_get_response_none(control_desc(head->type),
1537	head->type);
1538	} else if (head->type != expecting \|\| head->type == RDMA_CONTROL_ERROR) {
1539	error_setg(errp, "Was expecting a %s (%d) control message"
1540	", but got: %s (%d), length: %d",
1541	control_desc(expecting), expecting,
1542	control_desc(head->type), head->type, head->len);
1543	if (head->type == RDMA_CONTROL_ERROR) {
1544	rdma->received_error = true;
1545	}
1546	return -1;
1547	}
1548	if (head->len > RDMA_CONTROL_MAX_BUFFER - sizeof(*head)) {
1549	error_setg(errp, "too long length: %d", head->len);
1550	return -1;
1551	}
1552	if (sizeof(*head) + head->len != byte_len) {
1553	error_setg(errp, "Malformed length: %d byte_len %d",
1554	head->len, byte_len);
1555	return -1;
1556	}
1557
1558	return 0;
1559	}
1560
1561	/*
1562	* When a RECV work request has completed, the work request's
1563	* buffer is pointed at the header.
1564	*
1565	* This will advance the pointer to the data portion
1566	* of the control message of the work request's buffer that
1567	* was populated after the work request finished.
1568	*/
1569	static void qemu_rdma_move_header(RDMAContext *rdma, int idx,
1570	RDMAControlHeader *head)
1571	{
1572	rdma->wr_data[idx].control_len = head->len;
1573	rdma->wr_data[idx].control_curr =
1574	rdma->wr_data[idx].control + sizeof(RDMAControlHeader);
1575	}
1576
1577	/*
1578	* This is an 'atomic' high-level operation to deliver a single, unified
1579	* control-channel message.
1580	*
1581	* Additionally, if the user is expecting some kind of reply to this message,
1582	* they can request a 'resp' response message be filled in by posting an
1583	* additional work request on behalf of the user and waiting for an additional
1584	* completion.
1585	*
1586	* The extra (optional) response is used during registration to us from having
1587	* to perform an additional exchange of message just to provide a response by
1588	* instead piggy-backing on the acknowledgement.
1589	*/
1590	static int qemu_rdma_exchange_send(RDMAContext rdma, RDMAControlHeader head,
1591	uint8_t data, RDMAControlHeader resp,
1592	int *resp_idx,
1593	int (callback)(RDMAContext rdma,
1594	Error **errp),
1595	Error **errp)
1596	{
1597	int ret;
1598
1599	/*
1600	* Wait until the dest is ready before attempting to deliver the message
1601	* by waiting for a READY message.
1602	*/
1603	if (rdma->control_ready_expected) {
1604	RDMAControlHeader resp_ignored;
1605
1606	ret = qemu_rdma_exchange_get_response(rdma, &resp_ignored,
1607	RDMA_CONTROL_READY,
1608	RDMA_WRID_READY, errp);
1609	if (ret < 0) {
1610	return -1;
1611	}
1612	}
1613
1614	/*
1615	* If the user is expecting a response, post a WR in anticipation of it.
1616	*/
1617	if (resp) {
1618	ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_DATA, errp);
1619	if (ret < 0) {
1620	return -1;
1621	}
1622	}
1623
1624	/*
1625	* Post a WR to replace the one we just consumed for the READY message.
1626	*/
1627	ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY, errp);
1628	if (ret < 0) {
1629	return -1;
1630	}
1631
1632	/*
1633	* Deliver the control message that was requested.
1634	*/
1635	ret = qemu_rdma_post_send_control(rdma, data, head, errp);
1636
1637	if (ret < 0) {
1638	return -1;
1639	}
1640
1641	/*
1642	* If we're expecting a response, block and wait for it.
1643	*/
1644	if (resp) {
1645	if (callback) {
1646	trace_rdma_exchange_send_issue_callback();
1647	ret = callback(rdma, errp);
1648	if (ret < 0) {
1649	return -1;
1650	}
1651	}
1652
1653	trace_rdma_exchange_send_waiting(control_desc(resp->type));
1654	ret = qemu_rdma_exchange_get_response(rdma, resp,
1655	resp->type, RDMA_WRID_DATA,
1656	errp);
1657
1658	if (ret < 0) {
1659	return -1;
1660	}
1661
1662	qemu_rdma_move_header(rdma, RDMA_WRID_DATA, resp);
1663	if (resp_idx) {
1664	*resp_idx = RDMA_WRID_DATA;
1665	}
1666	trace_rdma_exchange_send_received(control_desc(resp->type));
1667	}
1668
1669	rdma->control_ready_expected = 1;
1670
1671	return 0;
1672	}
1673
1674	/*
1675	* This is an 'atomic' high-level operation to receive a single, unified
1676	* control-channel message.
1677	*/
1678	static int qemu_rdma_exchange_recv(RDMAContext rdma, RDMAControlHeader head,
1679	uint32_t expecting, Error **errp)
1680	{
1681	RDMAControlHeader ready = {
1682	.len = 0,
1683	.type = RDMA_CONTROL_READY,
1684	.repeat = 1,
1685	};
1686	int ret;
1687
1688	/*
1689	* Inform the source that we're ready to receive a message.
1690	*/
1691	ret = qemu_rdma_post_send_control(rdma, NULL, &ready, errp);
1692
1693	if (ret < 0) {
1694	return -1;
1695	}
1696
1697	/*
1698	* Block and wait for the message.
1699	*/
1700	ret = qemu_rdma_exchange_get_response(rdma, head,
1701	expecting, RDMA_WRID_READY, errp);
1702
1703	if (ret < 0) {
1704	return -1;
1705	}
1706
1707	qemu_rdma_move_header(rdma, RDMA_WRID_READY, head);
1708
1709	/*
1710	* Post a new RECV work request to replace the one we just consumed.
1711	*/
1712	ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY, errp);
1713	if (ret < 0) {
1714	return -1;
1715	}
1716
1717	return 0;
1718	}
1719
1720	/*
1721	* Write an actual chunk of memory using RDMA.
1722	*
1723	* If we're using dynamic registration on the dest-side, we have to
1724	* send a registration command first.
1725	*/
1726	static int qemu_rdma_write_one(RDMAContext *rdma,
1727	int current_index, uint64_t current_addr,
1728	uint64_t length, Error **errp)
1729	{
1730	struct ibv_sge sge;
1731	struct ibv_send_wr send_wr = { 0 };
1732	struct ibv_send_wr *bad_wr;
1733	int reg_result_idx, ret, count = 0;
1734	uint64_t chunk, chunks;
1735	uint64_t chunk_size = migrate_rdma_chunk_size();
1736	uint8_t chunk_start, chunk_end;
1737	RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
1738	RDMARegister reg;
1739	RDMARegisterResult *reg_result;
1740	RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
1741	RDMAControlHeader head = { .len = sizeof(RDMARegister),
1742	.type = RDMA_CONTROL_REGISTER_REQUEST,
1743	.repeat = 1,
1744	};
1745
1746	retry:
1747	sge.addr = (uintptr_t)(block->local_host_addr +
1748	(current_addr - block->offset));
1749	sge.length = length;
1750
1751	chunk = ram_chunk_index(block->local_host_addr,
1752	(uint8_t *)(uintptr_t)sge.addr);
1753	chunk_start = ram_chunk_start(block, chunk);
1754	chunks = length / chunk_size;
1755
1756	if (chunks && ((length % chunk_size) == 0)) {
1757	chunks--;
1758	}
1759
1760	trace_rdma_write_one_top(chunks + 1,
1761	(chunks + 1) * chunk_size / 1024 / 1024);
1762
1763	chunk_end = ram_chunk_end(block, chunk + chunks);
1764
1765
1766	while (test_bit(chunk, block->transit_bitmap)) {
1767	(void)count;
1768	trace_rdma_write_one_block(count++, current_index, chunk,
1769	sge.addr, length, rdma->nb_sent, block->nb_chunks);
1770
1771	ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
1772
1773	if (ret < 0) {
1774	error_setg(errp, "Failed to Wait for previous write to complete "
1775	"block %d chunk %" PRIu64
1776	" current %" PRIu64 " len %" PRIu64 " %d",
1777	current_index, chunk, sge.addr, length, rdma->nb_sent);
1778	return -1;
1779	}
1780	}
1781
1782	if (!rdma->pin_all) {
1783	if (!block->remote_keys[chunk]) {
1784	/*
1785	* This chunk has not yet been registered, so first check to see
1786	* if the entire chunk is zero. If so, tell the other size to
1787	* memset() + madvise() the entire chunk without RDMA.
1788	*/
1789
1790	if (buffer_is_zero((void *)(uintptr_t)sge.addr, length)) {
1791	RDMACompress comp = {
1792	.offset = current_addr,
1793	.value = 0,
1794	.block_idx = current_index,
1795	.length = length,
1796	};
1797
1798	head.len = sizeof(comp);
1799	head.type = RDMA_CONTROL_COMPRESS;
1800
1801	trace_rdma_write_one_zero(chunk, sge.length,
1802	current_index, current_addr);
1803
1804	compress_to_network(rdma, &comp);
1805	ret = qemu_rdma_exchange_send(rdma, &head,
1806	(uint8_t *) &comp, NULL, NULL, NULL, errp);
1807
1808	if (ret < 0) {
1809	return -1;
1810	}
1811
1812	/*
1813	* TODO: Here we are sending something, but we are not
1814	* accounting for anything transferred. The following is wrong:
1815	*
1816	* stat64_add(&mig_stats.rdma_bytes, sge.length);
1817	*
1818	* because we are using some kind of compression. I
1819	* would think that head.len would be the more similar
1820	* thing to a correct value.
1821	*/
1822	qatomic_add(&mig_stats.zero_pages,
1823	sge.length / qemu_target_page_size());
1824	return 1;
1825	}
1826
1827	/*
1828	* Otherwise, tell other side to register.
1829	*/
1830	reg.current_index = current_index;
1831	reg.current_addr = current_addr;
1832	reg.chunks = chunks;
1833
1834	trace_rdma_write_one_sendreg(chunk, sge.length, current_index,
1835	current_addr);
1836
1837	register_to_network(rdma, &reg);
1838	ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
1839	&resp, &reg_result_idx, NULL, errp);
1840	if (ret < 0) {
1841	return -1;
1842	}
1843
1844	/* try to overlap this single registration with the one we sent. */
1845	if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
1846	&sge.lkey, NULL, chunk,
1847	chunk_start, chunk_end)) {
1848	error_setg(errp, "cannot get lkey");
1849	return -1;
1850	}
1851
1852	reg_result = (RDMARegisterResult *)
1853	rdma->wr_data[reg_result_idx].control_curr;
1854
1855	network_to_result(reg_result);
1856
1857	trace_rdma_write_one_recvregres(block->remote_keys[chunk],
1858	reg_result->rkey, chunk);
1859
1860	block->remote_keys[chunk] = reg_result->rkey;
1861	block->remote_host_addr = reg_result->host_addr;
1862	} else {
1863	/* already registered before */
1864	if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
1865	&sge.lkey, NULL, chunk,
1866	chunk_start, chunk_end)) {
1867	error_setg(errp, "cannot get lkey!");
1868	return -1;
1869	}
1870	}
1871
1872	send_wr.wr.rdma.rkey = block->remote_keys[chunk];
1873	} else {
1874	send_wr.wr.rdma.rkey = block->remote_rkey;
1875
1876	if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
1877	&sge.lkey, NULL, chunk,
1878	chunk_start, chunk_end)) {
1879	error_setg(errp, "cannot get lkey!");
1880	return -1;
1881	}
1882	}
1883
1884	/*
1885	* Encode the ram block index and chunk within this wrid.
1886	* We will use this information at the time of completion
1887	* to figure out which bitmap to check against and then which
1888	* chunk in the bitmap to look for.
1889	*/
1890	send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
1891	current_index, chunk);
1892
1893	send_wr.opcode = IBV_WR_RDMA_WRITE;
1894	send_wr.send_flags = IBV_SEND_SIGNALED;
1895	send_wr.sg_list = &sge;
1896	send_wr.num_sge = 1;
1897	send_wr.wr.rdma.remote_addr = block->remote_host_addr +
1898	(current_addr - block->offset);
1899
1900	trace_rdma_write_one_post(chunk, sge.addr, send_wr.wr.rdma.remote_addr,
1901	sge.length);
1902
1903	/*
1904	* ibv_post_send() does not return negative error numbers,
1905	* per the specification they are positive - no idea why.
1906	*/
1907	ret = ibv_post_send(rdma->qp, &send_wr, &bad_wr);
1908
1909	if (ret == ENOMEM) {
1910	trace_rdma_write_one_queue_full();
1911	ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
1912	if (ret < 0) {
1913	error_setg(errp, "rdma migration: failed to make "
1914	"room in full send queue!");
1915	return -1;
1916	}
1917
1918	goto retry;
1919
1920	} else if (ret > 0) {
1921	error_setg_errno(errp, ret,
1922	"rdma migration: post rdma write failed");
1923	return -1;
1924	}
1925
1926	set_bit(chunk, block->transit_bitmap);
1927	qatomic_add(&mig_stats.normal_pages, sge.length / qemu_target_page_size());
1928	/*
1929	* We are adding to transferred the amount of data written, but no
1930	* overhead at all. I will assume that RDMA is magicaly and don't
1931	* need to transfer (at least) the addresses where it wants to
1932	* write the pages. Here it looks like it should be something
1933	* like:
1934	* sizeof(send_wr) + sge.length
1935	* but this being RDMA, who knows.
1936	*/
1937	qatomic_add(&mig_stats.rdma_bytes, sge.length);
1938	ram_transferred_add(sge.length);
1939	rdma->total_writes++;
1940
1941	return 0;
1942	}
1943
1944	/*
1945	* Push out any unwritten RDMA operations.
1946	*
1947	* We support sending out multiple chunks at the same time.
1948	* Not all of them need to get signaled in the completion queue.
1949	*/
1950	static int qemu_rdma_write_flush(RDMAContext rdma, Error *errp)
1951	{
1952	int ret;
1953
1954	if (!rdma->current_length) {
1955	return 0;
1956	}
1957
1958	ret = qemu_rdma_write_one(rdma, rdma->current_index, rdma->current_addr,
1959	rdma->current_length, errp);
1960
1961	if (ret < 0) {
1962	return -1;
1963	}
1964
1965	if (ret == 0) {
1966	rdma->nb_sent++;
1967	trace_rdma_write_flush(rdma->nb_sent);
1968	}
1969
1970	rdma->current_length = 0;
1971	rdma->current_addr = 0;
1972
1973	return 0;
1974	}
1975
1976	static inline bool qemu_rdma_buffer_mergeable(RDMAContext *rdma,
1977	uint64_t offset, uint64_t len)
1978	{
1979	RDMALocalBlock *block;
1980	uint8_t *host_addr;
1981	uint8_t *chunk_end;
1982
1983	if (rdma->current_index < 0) {
1984	return false;
1985	}
1986
1987	if (rdma->current_chunk < 0) {
1988	return false;
1989	}
1990
1991	block = &(rdma->local_ram_blocks.block[rdma->current_index]);
1992	host_addr = block->local_host_addr + (offset - block->offset);
1993	chunk_end = ram_chunk_end(block, rdma->current_chunk);
1994
1995	if (rdma->current_length == 0) {
1996	return false;
1997	}
1998
1999	/*
2000	* Only merge into chunk sequentially.
2001	*/
2002	if (offset != (rdma->current_addr + rdma->current_length)) {
2003	return false;
2004	}
2005
2006	if (offset < block->offset) {
2007	return false;
2008	}
2009
2010	if ((offset + len) > (block->offset + block->length)) {
2011	return false;
2012	}
2013
2014	if ((host_addr + len) > chunk_end) {
2015	return false;
2016	}
2017
2018	return true;
2019	}
2020
2021	/*
2022	* We're not actually writing here, but doing three things:
2023	*
2024	* 1. Identify the chunk the buffer belongs to.
2025	* 2. If the chunk is full or the buffer doesn't belong to the current
2026	* chunk, then start a new chunk and flush() the old chunk.
2027	* 3. To keep the hardware busy, we also group chunks into batches
2028	* and only require that a batch gets acknowledged in the completion
2029	* queue instead of each individual chunk.
2030	*/
2031	static int qemu_rdma_write(RDMAContext *rdma,
2032	uint64_t block_offset, uint64_t offset,
2033	uint64_t len, Error **errp)
2034	{
2035	uint64_t current_addr = block_offset + offset;
2036	uint64_t index = rdma->current_index;
2037	uint64_t chunk = rdma->current_chunk;
2038
2039	/* If we cannot merge it, we flush the current buffer first. */
2040	if (!qemu_rdma_buffer_mergeable(rdma, current_addr, len)) {
2041	if (qemu_rdma_write_flush(rdma, errp) < 0) {
2042	return -1;
2043	}
2044	rdma->current_length = 0;
2045	rdma->current_addr = current_addr;
2046
2047	qemu_rdma_search_ram_block(rdma, block_offset,
2048	offset, len, &index, &chunk);
2049	rdma->current_index = index;
2050	rdma->current_chunk = chunk;
2051	}
2052
2053	/* merge it */
2054	rdma->current_length += len;
2055
2056	/* flush it if buffer is too large */
2057	if (rdma->current_length >= rdma_merge_max()) {
2058	return qemu_rdma_write_flush(rdma, errp);
2059	}
2060
2061	return 0;
2062	}
2063
2064	static void qemu_rdma_cleanup(RDMAContext *rdma)
2065	{
2066	Error *err = NULL;
2067
2068	if (rdma->cm_id && rdma->connected) {
2069	if ((rdma->errored \|\|
2070	migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) &&
2071	!rdma->received_error) {
2072	RDMAControlHeader head = { .len = 0,
2073	.type = RDMA_CONTROL_ERROR,
2074	.repeat = 1,
2075	};
2076	warn_report("Early error. Sending error.");
2077	if (qemu_rdma_post_send_control(rdma, NULL, &head, &err) < 0) {
2078	warn_report_err(err);
2079	}
2080	}
2081
2082	rdma_disconnect(rdma->cm_id);
2083	trace_rdma_cleanup_disconnect();
2084	rdma->connected = false;
2085	}
2086
2087	if (rdma->channel) {
2088	qemu_set_fd_handler(rdma->channel->fd, NULL, NULL, NULL);
2089	}
2090	g_free(rdma->dest_blocks);
2091	rdma->dest_blocks = NULL;
2092
2093	for (int i = 0; i < RDMA_WRID_MAX; i++) {
2094	if (rdma->wr_data[i].control_mr) {
2095	rdma->total_registrations--;
2096	ibv_dereg_mr(rdma->wr_data[i].control_mr);
2097	}
2098	rdma->wr_data[i].control_mr = NULL;
2099	}
2100
2101	if (rdma->local_ram_blocks.block) {
2102	while (rdma->local_ram_blocks.nb_blocks) {
2103	rdma_delete_block(rdma, &rdma->local_ram_blocks.block[0]);
2104	}
2105	}
2106
2107	if (rdma->qp) {
2108	rdma_destroy_qp(rdma->cm_id);
2109	rdma->qp = NULL;
2110	}
2111	if (rdma->recv_cq) {
2112	ibv_destroy_cq(rdma->recv_cq);
2113	rdma->recv_cq = NULL;
2114	}
2115	if (rdma->send_cq) {
2116	ibv_destroy_cq(rdma->send_cq);
2117	rdma->send_cq = NULL;
2118	}
2119	if (rdma->recv_comp_channel) {
2120	ibv_destroy_comp_channel(rdma->recv_comp_channel);
2121	rdma->recv_comp_channel = NULL;
2122	}
2123	if (rdma->send_comp_channel) {
2124	ibv_destroy_comp_channel(rdma->send_comp_channel);
2125	rdma->send_comp_channel = NULL;
2126	}
2127	if (rdma->pd) {
2128	ibv_dealloc_pd(rdma->pd);
2129	rdma->pd = NULL;
2130	}
2131	if (rdma->cm_id) {
2132	rdma_destroy_id(rdma->cm_id);
2133	rdma->cm_id = NULL;
2134	}
2135
2136	/* the destination side, listen_id and channel is shared */
2137	if (rdma->listen_id) {
2138	if (!rdma->is_return_path) {
2139	rdma_destroy_id(rdma->listen_id);
2140	}
2141	rdma->listen_id = NULL;
2142
2143	if (rdma->channel) {
2144	if (!rdma->is_return_path) {
2145	rdma_destroy_event_channel(rdma->channel);
2146	}
2147	rdma->channel = NULL;
2148	}
2149	}
2150
2151	if (rdma->channel) {
2152	rdma_destroy_event_channel(rdma->channel);
2153	rdma->channel = NULL;
2154	}
2155	g_free(rdma->host);
2156	rdma->host = NULL;
2157	}
2158
2159
2160	static int qemu_rdma_source_init(RDMAContext rdma, bool pin_all, Error *errp)
2161	{
2162	int ret;
2163
2164	/*
2165	* Will be validated against destination's actual capabilities
2166	* after the connect() completes.
2167	*/
2168	rdma->pin_all = pin_all;
2169
2170	ret = qemu_rdma_resolve_host(rdma, errp);
2171	if (ret < 0) {
2172	goto err_rdma_source_init;
2173	}
2174
2175	ret = qemu_rdma_alloc_pd_cq(rdma, errp);
2176	if (ret < 0) {
2177	goto err_rdma_source_init;
2178	}
2179
2180	ret = qemu_rdma_alloc_qp(rdma);
2181	if (ret < 0) {
2182	error_setg(errp, "RDMA ERROR: rdma migration: error allocating qp!");
2183	goto err_rdma_source_init;
2184	}
2185
2186	qemu_rdma_init_ram_blocks(rdma);
2187
2188	/* Build the hash that maps from offset to RAMBlock */
2189	rdma->blockmap = g_hash_table_new(g_direct_hash, g_direct_equal);
2190	for (int i = 0; i < rdma->local_ram_blocks.nb_blocks; i++) {
2191	g_hash_table_insert(rdma->blockmap,
2192	(void *)(uintptr_t)rdma->local_ram_blocks.block[i].offset,
2193	&rdma->local_ram_blocks.block[i]);
2194	}
2195
2196	for (int i = 0; i < RDMA_WRID_MAX; i++) {
2197	ret = qemu_rdma_reg_control(rdma, i);
2198	if (ret < 0) {
2199	error_setg(errp, "RDMA ERROR: rdma migration: error "
2200	"registering %d control!", i);
2201	goto err_rdma_source_init;
2202	}
2203	}
2204
2205	return 0;
2206
2207	err_rdma_source_init:
2208	qemu_rdma_cleanup(rdma);
2209	return -1;
2210	}
2211
2212	static int qemu_get_cm_event_timeout(RDMAContext *rdma,
2213	struct rdma_cm_event **cm_event,
2214	long msec, Error **errp)
2215	{
2216	int ret;
2217	struct pollfd poll_fd = {
2218	.fd = rdma->channel->fd,
2219	.events = POLLIN,
2220	.revents = 0
2221	};
2222
2223	do {
2224	ret = poll(&poll_fd, 1, msec);
2225	} while (ret < 0 && errno == EINTR);
2226
2227	if (ret == 0) {
2228	error_setg(errp, "RDMA ERROR: poll cm event timeout");
2229	return -1;
2230	} else if (ret < 0) {
2231	error_setg_errno(errp, errno, "RDMA ERROR: failed to poll cm event");
2232	return -1;
2233	} else if (poll_fd.revents & POLLIN) {
2234	if (rdma_get_cm_event(rdma->channel, cm_event) < 0) {
2235	error_setg(errp, "RDMA ERROR: failed to get cm event");
2236	return -1;
2237	}
2238	return 0;
2239	} else {
2240	error_setg(errp, "RDMA ERROR: no POLLIN event, revent=%x",
2241	poll_fd.revents);
2242	return -1;
2243	}
2244	}
2245
2246	static int qemu_rdma_connect(RDMAContext *rdma, bool return_path,
2247	Error **errp)
2248	{
2249	RDMACapabilities cap = {
2250	.version = RDMA_CONTROL_VERSION_CURRENT,
2251	.flags = 0,
2252	};
2253	struct rdma_conn_param conn_param = { .initiator_depth = 2,
2254	.retry_count = 5,
2255	.private_data = &cap,
2256	.private_data_len = sizeof(cap),
2257	};
2258	struct rdma_cm_event *cm_event;
2259	int ret;
2260
2261	/*
2262	* Only negotiate the capability with destination if the user
2263	* on the source first requested the capability.
2264	*/
2265	if (rdma->pin_all) {
2266	trace_rdma_connect_pin_all_requested();
2267	cap.flags \|= RDMA_CAPABILITY_PIN_ALL;
2268	}
2269
2270	caps_to_network(&cap);
2271
2272	ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY, errp);
2273	if (ret < 0) {
2274	goto err_rdma_source_connect;
2275	}
2276
2277	ret = rdma_connect(rdma->cm_id, &conn_param);
2278	if (ret < 0) {
2279	error_setg_errno(errp, errno,
2280	"RDMA ERROR: connecting to destination!");
2281	goto err_rdma_source_connect;
2282	}
2283
2284	if (return_path) {
2285	ret = qemu_get_cm_event_timeout(rdma, &cm_event, 5000, errp);
2286	} else {
2287	ret = rdma_get_cm_event(rdma->channel, &cm_event);
2288	if (ret < 0) {
2289	error_setg_errno(errp, errno,
2290	"RDMA ERROR: failed to get cm event");
2291	}
2292	}
2293	if (ret < 0) {
2294	goto err_rdma_source_connect;
2295	}
2296
2297	if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
2298	error_setg(errp, "RDMA ERROR: connecting to destination!");
2299	rdma_ack_cm_event(cm_event);
2300	goto err_rdma_source_connect;
2301	}
2302	rdma->connected = true;
2303
2304	memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
2305	network_to_caps(&cap);
2306
2307	/*
2308	* Verify that the requested capabilities are supported by the destination
2309	* and disable them otherwise.
2310	*/
2311	if (rdma->pin_all && !(cap.flags & RDMA_CAPABILITY_PIN_ALL)) {
2312	warn_report("RDMA: Server cannot support pinning all memory. "
2313	"Will register memory dynamically.");
2314	rdma->pin_all = false;
2315	}
2316
2317	trace_rdma_connect_pin_all_outcome(rdma->pin_all);
2318
2319	rdma_ack_cm_event(cm_event);
2320
2321	rdma->control_ready_expected = 1;
2322	rdma->nb_sent = 0;
2323	return 0;
2324
2325	err_rdma_source_connect:
2326	qemu_rdma_cleanup(rdma);
2327	return -1;
2328	}
2329
2330	static int qemu_rdma_dest_init(RDMAContext rdma, Error *errp)
2331	{
2332	int ret;
2333	struct rdma_cm_id *listen_id;
2334	char ip[40] = "unknown";
2335	struct rdma_addrinfo res, e;
2336	char port_str[16];
2337	int reuse = 1;
2338
2339	for (int i = 0; i < RDMA_WRID_MAX; i++) {
2340	rdma->wr_data[i].control_len = 0;
2341	rdma->wr_data[i].control_curr = NULL;
2342	}
2343
2344	if (!rdma->host \|\| !rdma->host[0]) {
2345	error_setg(errp, "RDMA ERROR: RDMA host is not set!");
2346	rdma->errored = true;
2347	return -1;
2348	}
2349	/* create CM channel */
2350	rdma->channel = rdma_create_event_channel();
2351	if (!rdma->channel) {
2352	error_setg(errp, "RDMA ERROR: could not create rdma event channel");
2353	rdma->errored = true;
2354	return -1;
2355	}
2356
2357	/* create CM id */
2358	ret = rdma_create_id(rdma->channel, &listen_id, NULL, RDMA_PS_TCP);
2359	if (ret < 0) {
2360	error_setg(errp, "RDMA ERROR: could not create cm_id!");
2361	goto err_dest_init_create_listen_id;
2362	}
2363
2364	snprintf(port_str, 16, "%d", rdma->port);
2365	port_str[15] = '\0';
2366
2367	ret = rdma_getaddrinfo(rdma->host, port_str, NULL, &res);
2368	if (ret) {
2369	error_setg(errp, "RDMA ERROR: could not rdma_getaddrinfo address %s",
2370	rdma->host);
2371	goto err_dest_init_bind_addr;
2372	}
2373
2374	ret = rdma_set_option(listen_id, RDMA_OPTION_ID, RDMA_OPTION_ID_REUSEADDR,
2375	&reuse, sizeof reuse);
2376	if (ret < 0) {
2377	error_setg(errp, "RDMA ERROR: Error: could not set REUSEADDR option");
2378	goto err_dest_init_bind_addr;
2379	}
2380
2381	/* Try all addresses */
2382	for (e = res; e != NULL; e = e->ai_next) {
2383
2384	inet_ntop(e->ai_family,
2385	&((struct sockaddr_in *) e->ai_dst_addr)->sin_addr, ip, sizeof ip);
2386	trace_rdma_dest_init_trying(rdma->host, ip);
2387	ret = rdma_bind_addr(listen_id, e->ai_dst_addr);
2388	if (ret < 0) {
2389	continue;
2390	}
2391	break;
2392	}
2393
2394	rdma_freeaddrinfo(res);
2395	if (!e) {
2396	error_setg(errp, "RDMA ERROR: Error: could not rdma_bind_addr!");
2397	goto err_dest_init_bind_addr;
2398	}
2399
2400	rdma->listen_id = listen_id;
2401	qemu_rdma_dump_gid("dest_init", listen_id);
2402	return 0;
2403
2404	err_dest_init_bind_addr:
2405	rdma_destroy_id(listen_id);
2406	err_dest_init_create_listen_id:
2407	rdma_destroy_event_channel(rdma->channel);
2408	rdma->channel = NULL;
2409	rdma->errored = true;
2410	return -1;
2411
2412	}
2413
2414	static void qemu_rdma_return_path_dest_init(RDMAContext *rdma_return_path,
2415	RDMAContext *rdma)
2416	{
2417	for (int i = 0; i < RDMA_WRID_MAX; i++) {
2418	rdma_return_path->wr_data[i].control_len = 0;
2419	rdma_return_path->wr_data[i].control_curr = NULL;
2420	}
2421
2422	/the CM channel and CM id is shared/
2423	rdma_return_path->channel = rdma->channel;
2424	rdma_return_path->listen_id = rdma->listen_id;
2425
2426	rdma->return_path = rdma_return_path;
2427	rdma_return_path->return_path = rdma;
2428	rdma_return_path->is_return_path = true;
2429	}
2430
2431	static RDMAContext qemu_rdma_data_init(InetSocketAddress saddr, Error **errp)
2432	{
2433	RDMAContext *rdma = NULL;
2434
2435	rdma = g_new0(RDMAContext, 1);
2436	rdma->current_index = -1;
2437	rdma->current_chunk = -1;
2438
2439	rdma->host = g_strdup(saddr->host);
2440	rdma->port = atoi(saddr->port);
2441	return rdma;
2442	}
2443
2444	/*
2445	* QEMUFile interface to the control channel.
2446	* SEND messages for control only.
2447	* VM's ram is handled with regular RDMA messages.
2448	*/
2449	static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
2450	const struct iovec *iov,
2451	size_t niov,
2452	int *fds,
2453	size_t nfds,
2454	int flags,
2455	Error **errp)
2456	{
2457	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2458	RDMAContext *rdma;
2459	int ret;
2460	ssize_t done = 0;
2461	size_t len;
2462
2463	RCU_READ_LOCK_GUARD();
2464	rdma = qatomic_rcu_read(&rioc->rdmaout);
2465
2466	if (!rdma) {
2467	error_setg(errp, "RDMA control channel output is not set");
2468	return -1;
2469	}
2470
2471	if (rdma->errored) {
2472	error_setg(errp,
2473	"RDMA is in an error state waiting migration to abort!");
2474	return -1;
2475	}
2476
2477	/*
2478	* Push out any writes that
2479	* we're queued up for VM's ram.
2480	*/
2481	ret = qemu_rdma_write_flush(rdma, errp);
2482	if (ret < 0) {
2483	rdma->errored = true;
2484	return -1;
2485	}
2486
2487	for (int i = 0; i < niov; i++) {
2488	size_t remaining = iov[i].iov_len;
2489	uint8_t * data = (void *)iov[i].iov_base;
2490	while (remaining) {
2491	RDMAControlHeader head = {};
2492
2493	len = MIN(remaining, RDMA_SEND_INCREMENT);
2494	remaining -= len;
2495
2496	head.len = len;
2497	head.type = RDMA_CONTROL_QEMU_FILE;
2498
2499	ret = qemu_rdma_exchange_send(rdma, &head,
2500	data, NULL, NULL, NULL, errp);
2501
2502	if (ret < 0) {
2503	rdma->errored = true;
2504	return -1;
2505	}
2506
2507	data += len;
2508	done += len;
2509	}
2510	}
2511
2512	return done;
2513	}
2514
2515	static size_t qemu_rdma_fill(RDMAContext rdma, uint8_t buf,
2516	size_t size, int idx)
2517	{
2518	size_t len = 0;
2519
2520	if (rdma->wr_data[idx].control_len) {
2521	trace_rdma_fill(rdma->wr_data[idx].control_len, size);
2522
2523	len = MIN(size, rdma->wr_data[idx].control_len);
2524	memcpy(buf, rdma->wr_data[idx].control_curr, len);
2525	rdma->wr_data[idx].control_curr += len;
2526	rdma->wr_data[idx].control_len -= len;
2527	}
2528
2529	return len;
2530	}
2531
2532	/*
2533	* QEMUFile interface to the control channel.
2534	* RDMA links don't use bytestreams, so we have to
2535	* return bytes to QEMUFile opportunistically.
2536	*/
2537	static ssize_t qio_channel_rdma_readv(QIOChannel *ioc,
2538	const struct iovec *iov,
2539	size_t niov,
2540	int **fds,
2541	size_t *nfds,
2542	int flags,
2543	Error **errp)
2544	{
2545	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2546	RDMAContext *rdma;
2547	RDMAControlHeader head;
2548	int ret;
2549	ssize_t done = 0;
2550	size_t len;
2551
2552	RCU_READ_LOCK_GUARD();
2553	rdma = qatomic_rcu_read(&rioc->rdmain);
2554
2555	if (!rdma) {
2556	error_setg(errp, "RDMA control channel input is not set");
2557	return -1;
2558	}
2559
2560	if (rdma->errored) {
2561	error_setg(errp,
2562	"RDMA is in an error state waiting migration to abort!");
2563	return -1;
2564	}
2565
2566	for (int i = 0; i < niov; i++) {
2567	size_t want = iov[i].iov_len;
2568	uint8_t data = (void )iov[i].iov_base;
2569
2570	/*
2571	* First, we hold on to the last SEND message we
2572	* were given and dish out the bytes until we run
2573	* out of bytes.
2574	*/
2575	len = qemu_rdma_fill(rdma, data, want, 0);
2576	done += len;
2577	want -= len;
2578	/* Got what we needed, so go to next iovec */
2579	if (want == 0) {
2580	continue;
2581	}
2582
2583	/* If we got any data so far, then don't wait
2584	* for more, just return what we have */
2585	if (done > 0) {
2586	break;
2587	}
2588
2589
2590	/* We've got nothing at all, so lets wait for
2591	* more to arrive
2592	*/
2593	ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_QEMU_FILE,
2594	errp);
2595
2596	if (ret < 0) {
2597	rdma->errored = true;
2598	return -1;
2599	}
2600
2601	/*
2602	* SEND was received with new bytes, now try again.
2603	*/
2604	len = qemu_rdma_fill(rdma, data, want, 0);
2605	done += len;
2606	want -= len;
2607
2608	/* Still didn't get enough, so lets just return */
2609	if (want) {
2610	if (done == 0) {
2611	return QIO_CHANNEL_ERR_BLOCK;
2612	} else {
2613	break;
2614	}
2615	}
2616	}
2617	return done;
2618	}
2619
2620	/*
2621	* Block until all the outstanding chunks have been delivered by the hardware.
2622	*/
2623	static int qemu_rdma_drain_cq(RDMAContext *rdma)
2624	{
2625	Error *err = NULL;
2626
2627	if (qemu_rdma_write_flush(rdma, &err) < 0) {
2628	error_report_err(err);
2629	return -1;
2630	}
2631
2632	while (rdma->nb_sent) {
2633	if (qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL) < 0) {
2634	error_report("rdma migration: complete polling error!");
2635	return -1;
2636	}
2637	}
2638
2639	return 0;
2640	}
2641
2642
2643	static int qio_channel_rdma_set_blocking(QIOChannel *ioc,
2644	bool blocking,
2645	Error **errp)
2646	{
2647	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2648	/* XXX we should make readv/writev actually honour this :-) */
2649	rioc->blocking = blocking;
2650	return 0;
2651	}
2652
2653
2654	typedef struct QIOChannelRDMASource QIOChannelRDMASource;
2655	struct QIOChannelRDMASource {
2656	GSource parent;
2657	QIOChannelRDMA *rioc;
2658	GIOCondition condition;
2659	};
2660
2661	static gboolean
2662	qio_channel_rdma_source_prepare(GSource *source,
2663	gint *timeout)
2664	{
2665	QIOChannelRDMASource rsource = (QIOChannelRDMASource )source;
2666	RDMAContext *rdma;
2667	GIOCondition cond = 0;
2668	*timeout = -1;
2669
2670	RCU_READ_LOCK_GUARD();
2671	if (rsource->condition == G_IO_IN) {
2672	rdma = qatomic_rcu_read(&rsource->rioc->rdmain);
2673	} else {
2674	rdma = qatomic_rcu_read(&rsource->rioc->rdmaout);
2675	}
2676
2677	if (!rdma) {
2678	error_report("RDMAContext is NULL when prepare Gsource");
2679	return FALSE;
2680	}
2681
2682	if (rdma->wr_data[0].control_len) {
2683	cond \|= G_IO_IN;
2684	}
2685	cond \|= G_IO_OUT;
2686
2687	return cond & rsource->condition;
2688	}
2689
2690	static gboolean
2691	qio_channel_rdma_source_check(GSource *source)
2692	{
2693	QIOChannelRDMASource rsource = (QIOChannelRDMASource )source;
2694	RDMAContext *rdma;
2695	GIOCondition cond = 0;
2696
2697	RCU_READ_LOCK_GUARD();
2698	if (rsource->condition == G_IO_IN) {
2699	rdma = qatomic_rcu_read(&rsource->rioc->rdmain);
2700	} else {
2701	rdma = qatomic_rcu_read(&rsource->rioc->rdmaout);
2702	}
2703
2704	if (!rdma) {
2705	error_report("RDMAContext is NULL when check Gsource");
2706	return FALSE;
2707	}
2708
2709	if (rdma->wr_data[0].control_len) {
2710	cond \|= G_IO_IN;
2711	}
2712	cond \|= G_IO_OUT;
2713
2714	return cond & rsource->condition;
2715	}
2716
2717	static gboolean
2718	qio_channel_rdma_source_dispatch(GSource *source,
2719	GSourceFunc callback,
2720	gpointer user_data)
2721	{
2722	QIOChannelFunc func = (QIOChannelFunc)callback;
2723	QIOChannelRDMASource rsource = (QIOChannelRDMASource )source;
2724	RDMAContext *rdma;
2725	GIOCondition cond = 0;
2726
2727	RCU_READ_LOCK_GUARD();
2728	if (rsource->condition == G_IO_IN) {
2729	rdma = qatomic_rcu_read(&rsource->rioc->rdmain);
2730	} else {
2731	rdma = qatomic_rcu_read(&rsource->rioc->rdmaout);
2732	}
2733
2734	if (!rdma) {
2735	error_report("RDMAContext is NULL when dispatch Gsource");
2736	return FALSE;
2737	}
2738
2739	if (rdma->wr_data[0].control_len) {
2740	cond \|= G_IO_IN;
2741	}
2742	cond \|= G_IO_OUT;
2743
2744	return (*func)(QIO_CHANNEL(rsource->rioc),
2745	(cond & rsource->condition),
2746	user_data);
2747	}
2748
2749	static void
2750	qio_channel_rdma_source_finalize(GSource *source)
2751	{
2752	QIOChannelRDMASource ssource = (QIOChannelRDMASource )source;
2753
2754	object_unref(OBJECT(ssource->rioc));
2755	}
2756
2757	static GSourceFuncs qio_channel_rdma_source_funcs = {
2758	qio_channel_rdma_source_prepare,
2759	qio_channel_rdma_source_check,
2760	qio_channel_rdma_source_dispatch,
2761	qio_channel_rdma_source_finalize
2762	};
2763
2764	static GSource qio_channel_rdma_create_watch(QIOChannel ioc,
2765	GIOCondition condition)
2766	{
2767	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2768	QIOChannelRDMASource *ssource;
2769	GSource *source;
2770
2771	source = g_source_new(&qio_channel_rdma_source_funcs,
2772	sizeof(QIOChannelRDMASource));
2773	ssource = (QIOChannelRDMASource *)source;
2774
2775	ssource->rioc = rioc;
2776	object_ref(OBJECT(rioc));
2777
2778	ssource->condition = condition;
2779
2780	return source;
2781	}
2782
2783	static void qio_channel_rdma_set_aio_fd_handler(QIOChannel *ioc,
2784	AioContext *read_ctx,
2785	IOHandler *io_read,
2786	AioContext *write_ctx,
2787	IOHandler *io_write,
2788	void *opaque)
2789	{
2790	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2791	if (io_read) {
2792	aio_set_fd_handler(read_ctx, rioc->rdmain->recv_comp_channel->fd,
2793	io_read, io_write, NULL, NULL, opaque);
2794	aio_set_fd_handler(read_ctx, rioc->rdmain->send_comp_channel->fd,
2795	io_read, io_write, NULL, NULL, opaque);
2796	} else {
2797	aio_set_fd_handler(write_ctx, rioc->rdmaout->recv_comp_channel->fd,
2798	io_read, io_write, NULL, NULL, opaque);
2799	aio_set_fd_handler(write_ctx, rioc->rdmaout->send_comp_channel->fd,
2800	io_read, io_write, NULL, NULL, opaque);
2801	}
2802	}
2803
2804	struct rdma_close_rcu {
2805	struct rcu_head rcu;
2806	RDMAContext *rdmain;
2807	RDMAContext *rdmaout;
2808	};
2809
2810	/* callback from qio_channel_rdma_close via call_rcu */
2811	static void qio_channel_rdma_close_rcu(struct rdma_close_rcu *rcu)
2812	{
2813	if (rcu->rdmain) {
2814	qemu_rdma_cleanup(rcu->rdmain);
2815	}
2816
2817	if (rcu->rdmaout) {
2818	qemu_rdma_cleanup(rcu->rdmaout);
2819	}
2820
2821	g_free(rcu->rdmain);
2822	g_free(rcu->rdmaout);
2823	g_free(rcu);
2824	}
2825
2826	static int qio_channel_rdma_close(QIOChannel *ioc,
2827	Error **errp)
2828	{
2829	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2830	RDMAContext rdmain, rdmaout;
2831	struct rdma_close_rcu *rcu = g_new(struct rdma_close_rcu, 1);
2832
2833	trace_rdma_close();
2834
2835	rdmain = rioc->rdmain;
2836	if (rdmain) {
2837	qatomic_rcu_set(&rioc->rdmain, NULL);
2838	}
2839
2840	rdmaout = rioc->rdmaout;
2841	if (rdmaout) {
2842	qatomic_rcu_set(&rioc->rdmaout, NULL);
2843	}
2844
2845	rcu->rdmain = rdmain;
2846	rcu->rdmaout = rdmaout;
2847	call_rcu(rcu, qio_channel_rdma_close_rcu, rcu);
2848
2849	return 0;
2850	}
2851
2852	static int
2853	qio_channel_rdma_shutdown(QIOChannel *ioc,
2854	QIOChannelShutdown how,
2855	Error **errp)
2856	{
2857	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
2858	RDMAContext rdmain, rdmaout;
2859
2860	RCU_READ_LOCK_GUARD();
2861
2862	rdmain = qatomic_rcu_read(&rioc->rdmain);
2863	rdmaout = qatomic_rcu_read(&rioc->rdmain);
2864
2865	switch (how) {
2866	case QIO_CHANNEL_SHUTDOWN_READ:
2867	if (rdmain) {
2868	rdmain->errored = true;
2869	}
2870	break;
2871	case QIO_CHANNEL_SHUTDOWN_WRITE:
2872	if (rdmaout) {
2873	rdmaout->errored = true;
2874	}
2875	break;
2876	case QIO_CHANNEL_SHUTDOWN_BOTH:
2877	default:
2878	if (rdmain) {
2879	rdmain->errored = true;
2880	}
2881	if (rdmaout) {
2882	rdmaout->errored = true;
2883	}
2884	break;
2885	}
2886
2887	return 0;
2888	}
2889
2890	/*
2891	* Parameters:
2892	* @offset == 0 :
2893	* This means that 'block_offset' is a full virtual address that does not
2894	* belong to a RAMBlock of the virtual machine and instead
2895	* represents a private malloc'd memory area that the caller wishes to
2896	* transfer.
2897	*
2898	* @offset != 0 :
2899	* Offset is an offset to be added to block_offset and used
2900	* to also lookup the corresponding RAMBlock.
2901	*
2902	* @size : Number of bytes to transfer
2903	*
2904	* @pages_sent : User-specificed pointer to indicate how many pages were
2905	* sent. Usually, this will not be more than a few bytes of
2906	* the protocol because most transfers are sent asynchronously.
2907	*/
2908	static int qemu_rdma_save_page(QEMUFile *f, ram_addr_t block_offset,
2909	ram_addr_t offset, size_t size)
2910	{
2911	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
2912	Error *err = NULL;
2913	RDMAContext *rdma;
2914	int ret;
2915
2916	RCU_READ_LOCK_GUARD();
2917	rdma = qatomic_rcu_read(&rioc->rdmaout);
2918
2919	if (!rdma) {
2920	return -1;
2921	}
2922
2923	if (rdma_errored(rdma)) {
2924	return -1;
2925	}
2926
2927	qemu_fflush(f);
2928
2929	/*
2930	* Add this page to the current 'chunk'. If the chunk
2931	* is full, or the page doesn't belong to the current chunk,
2932	* an actual RDMA write will occur and a new chunk will be formed.
2933	*/
2934	ret = qemu_rdma_write(rdma, block_offset, offset, size, &err);
2935	if (ret < 0) {
2936	error_report_err(err);
2937	goto err;
2938	}
2939
2940	/*
2941	* Drain the Completion Queue if possible, but do not block,
2942	* just poll.
2943	*
2944	* If nothing to poll, the end of the iteration will do this
2945	* again to make sure we don't overflow the request queue.
2946	*/
2947	while (1) {
2948	uint64_t wr_id, wr_id_in;
2949	ret = qemu_rdma_poll(rdma, rdma->recv_cq, &wr_id_in, NULL);
2950
2951	if (ret < 0) {
2952	error_report("rdma migration: polling error");
2953	goto err;
2954	}
2955
2956	wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
2957
2958	if (wr_id == RDMA_WRID_NONE) {
2959	break;
2960	}
2961	}
2962
2963	while (1) {
2964	uint64_t wr_id, wr_id_in;
2965	ret = qemu_rdma_poll(rdma, rdma->send_cq, &wr_id_in, NULL);
2966
2967	if (ret < 0) {
2968	error_report("rdma migration: polling error");
2969	goto err;
2970	}
2971
2972	wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
2973
2974	if (wr_id == RDMA_WRID_NONE) {
2975	break;
2976	}
2977	}
2978
2979	return RAM_SAVE_CONTROL_DELAYED;
2980
2981	err:
2982	rdma->errored = true;
2983	return -1;
2984	}
2985
2986	int rdma_control_save_page(QEMUFile *f, ram_addr_t block_offset,
2987	ram_addr_t offset, size_t size)
2988	{
2989	assert(migrate_rdma());
2990
2991	int ret = qemu_rdma_save_page(f, block_offset, offset, size);
2992
2993	if (ret != RAM_SAVE_CONTROL_DELAYED) {
2994	if (ret < 0) {
2995	qemu_file_set_error(f, ret);
2996	}
2997	}
2998	return ret;
2999	}
3000
3001	static void rdma_accept_incoming_migration(void *opaque);
3002
3003	static void rdma_cm_poll_handler(void *opaque)
3004	{
3005	RDMAContext *rdma = opaque;
3006	struct rdma_cm_event *cm_event;
3007	MigrationIncomingState *mis = migration_incoming_get_current();
3008
3009	if (rdma_get_cm_event(rdma->channel, &cm_event) < 0) {
3010	error_report("get_cm_event failed %d", errno);
3011	return;
3012	}
3013
3014	if (cm_event->event == RDMA_CM_EVENT_DISCONNECTED \|\|
3015	cm_event->event == RDMA_CM_EVENT_DEVICE_REMOVAL) {
3016	if (!rdma->errored &&
3017	migration_incoming_get_current()->state !=
3018	MIGRATION_STATUS_COMPLETED) {
3019	error_report("receive cm event, cm event is %d", cm_event->event);
3020	rdma->errored = true;
3021	if (rdma->return_path) {
3022	rdma->return_path->errored = true;
3023	}
3024	}
3025	rdma_ack_cm_event(cm_event);
3026	if (mis->loadvm_co) {
3027	qemu_coroutine_enter(mis->loadvm_co);
3028	}
3029	return;
3030	}
3031	rdma_ack_cm_event(cm_event);
3032	}
3033
3034	static int qemu_rdma_accept(RDMAContext *rdma)
3035	{
3036	Error *err = NULL;
3037	RDMACapabilities cap;
3038	struct rdma_conn_param conn_param = {
3039	.responder_resources = 2,
3040	.private_data = &cap,
3041	.private_data_len = sizeof(cap),
3042	};
3043	RDMAContext *rdma_return_path = NULL;
3044	g_autoptr(InetSocketAddress) isock = g_new0(InetSocketAddress, 1);
3045	struct rdma_cm_event *cm_event;
3046	struct ibv_context *verbs;
3047	int ret;
3048
3049	ret = rdma_get_cm_event(rdma->channel, &cm_event);
3050	if (ret < 0) {
3051	goto err_rdma_dest_wait;
3052	}
3053
3054	if (cm_event->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
3055	rdma_ack_cm_event(cm_event);
3056	goto err_rdma_dest_wait;
3057	}
3058
3059	isock->host = g_strdup(rdma->host);
3060	isock->port = g_strdup_printf("%d", rdma->port);
3061
3062	/*
3063	* initialize the RDMAContext for return path for postcopy after first
3064	* connection request reached.
3065	*/
3066	if ((migrate_postcopy() \|\| migrate_return_path())
3067	&& !rdma->is_return_path) {
3068	rdma_return_path = qemu_rdma_data_init(isock, NULL);
3069	if (rdma_return_path == NULL) {
3070	rdma_ack_cm_event(cm_event);
3071	goto err_rdma_dest_wait;
3072	}
3073
3074	qemu_rdma_return_path_dest_init(rdma_return_path, rdma);
3075	}
3076
3077	memcpy(&cap, cm_event->param.conn.private_data, sizeof(cap));
3078
3079	network_to_caps(&cap);
3080
3081	if (cap.version < 1 \|\| cap.version > RDMA_CONTROL_VERSION_CURRENT) {
3082	error_report("Unknown source RDMA version: %d, bailing...",
3083	cap.version);
3084	rdma_ack_cm_event(cm_event);
3085	goto err_rdma_dest_wait;
3086	}
3087
3088	/*
3089	* Respond with only the capabilities this version of QEMU knows about.
3090	*/
3091	cap.flags &= known_capabilities;
3092
3093	/*
3094	* Enable the ones that we do know about.
3095	* Add other checks here as new ones are introduced.
3096	*/
3097	if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
3098	rdma->pin_all = true;
3099	}
3100
3101	rdma->cm_id = cm_event->id;
3102	verbs = cm_event->id->verbs;
3103
3104	rdma_ack_cm_event(cm_event);
3105
3106	trace_rdma_accept_pin_state(rdma->pin_all);
3107
3108	caps_to_network(&cap);
3109
3110	trace_rdma_accept_pin_verbsc(verbs);
3111
3112	if (!rdma->verbs) {
3113	rdma->verbs = verbs;
3114	} else if (rdma->verbs != verbs) {
3115	error_report("ibv context not matching %p, %p!", rdma->verbs,
3116	verbs);
3117	goto err_rdma_dest_wait;
3118	}
3119
3120	qemu_rdma_dump_id("dest_init", verbs);
3121
3122	ret = qemu_rdma_alloc_pd_cq(rdma, &err);
3123	if (ret < 0) {
3124	error_report_err(err);
3125	goto err_rdma_dest_wait;
3126	}
3127
3128	ret = qemu_rdma_alloc_qp(rdma);
3129	if (ret < 0) {
3130	error_report("rdma migration: error allocating qp!");
3131	goto err_rdma_dest_wait;
3132	}
3133
3134	qemu_rdma_init_ram_blocks(rdma);
3135
3136	for (int i = 0; i < RDMA_WRID_MAX; i++) {
3137	ret = qemu_rdma_reg_control(rdma, i);
3138	if (ret < 0) {
3139	error_report("rdma: error registering %d control", i);
3140	goto err_rdma_dest_wait;
3141	}
3142	}
3143
3144	/* Accept the second connection request for return path */
3145	if ((migrate_postcopy() \|\| migrate_return_path())
3146	&& !rdma->is_return_path) {
3147	qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
3148	NULL,
3149	(void *)(intptr_t)rdma->return_path);
3150	} else {
3151	qemu_set_fd_handler(rdma->channel->fd, rdma_cm_poll_handler,
3152	NULL, rdma);
3153	}
3154
3155	ret = rdma_accept(rdma->cm_id, &conn_param);
3156	if (ret < 0) {
3157	error_report("rdma_accept failed");
3158	goto err_rdma_dest_wait;
3159	}
3160
3161	ret = rdma_get_cm_event(rdma->channel, &cm_event);
3162	if (ret < 0) {
3163	error_report("rdma_accept get_cm_event failed");
3164	goto err_rdma_dest_wait;
3165	}
3166
3167	if (cm_event->event != RDMA_CM_EVENT_ESTABLISHED) {
3168	error_report("rdma_accept not event established");
3169	rdma_ack_cm_event(cm_event);
3170	goto err_rdma_dest_wait;
3171	}
3172
3173	rdma_ack_cm_event(cm_event);
3174	rdma->connected = true;
3175
3176	ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY, &err);
3177	if (ret < 0) {
3178	error_report_err(err);
3179	goto err_rdma_dest_wait;
3180	}
3181
3182	qemu_rdma_dump_gid("dest_connect", rdma->cm_id);
3183
3184	return 0;
3185
3186	err_rdma_dest_wait:
3187	rdma->errored = true;
3188	qemu_rdma_cleanup(rdma);
3189	g_free(rdma_return_path);
3190	return -1;
3191	}
3192
3193	static int dest_ram_sort_func(const void a, const void b)
3194	{
3195	unsigned int a_index = ((const RDMALocalBlock *)a)->src_index;
3196	unsigned int b_index = ((const RDMALocalBlock *)b)->src_index;
3197
3198	return (a_index < b_index) ? -1 : (a_index != b_index);
3199	}
3200
3201	static bool rdma_compress_range_check(RDMALocalBlock *block,
3202	RDMACompress *comp)
3203	{
3204	uint64_t block_end = block->offset + block->length;
3205	uint64_t comp_end;
3206
3207	if (uadd64_overflow(comp->offset, comp->length, &comp_end)) {
3208	goto fail;
3209	}
3210
3211	if (comp->offset < block->offset \|\| comp_end > block_end) {
3212	goto fail;
3213	}
3214
3215	return true;
3216	fail:
3217	error_report("%s: compress request range outside range"
3218	" (block=%s, offset=%"PRIu64", length=%"PRIu64")",
3219	__func__, block->block_name, comp->offset, comp->length);
3220	return false;
3221	}
3222
3223	/*
3224	* During each iteration of the migration, we listen for instructions
3225	* by the source VM to perform dynamic page registrations before they
3226	* can perform RDMA operations.
3227	*
3228	* We respond with the 'rkey'.
3229	*
3230	* Keep doing this until the source tells us to stop.
3231	*/
3232	int rdma_registration_handle(QEMUFile *f)
3233	{
3234	RDMAControlHeader reg_resp = { .len = sizeof(RDMARegisterResult),
3235	.type = RDMA_CONTROL_REGISTER_RESULT,
3236	.repeat = 0,
3237	};
3238	RDMAControlHeader blocks = { .type = RDMA_CONTROL_RAM_BLOCKS_RESULT,
3239	.repeat = 1 };
3240	QIOChannelRDMA *rioc;
3241	Error *err = NULL;
3242	RDMAContext *rdma;
3243	RDMALocalBlocks *local;
3244	RDMAControlHeader head;
3245	RDMARegister reg, registers;
3246	RDMACompress *comp;
3247	RDMARegisterResult *reg_result;
3248	static RDMARegisterResult results[RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE];
3249	RDMALocalBlock *block;
3250	void *host_addr;
3251	int ret;
3252	int idx = 0;
3253
3254	if (!migrate_rdma()) {
3255	return 0;
3256	}
3257
3258	RCU_READ_LOCK_GUARD();
3259	rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3260	rdma = qatomic_rcu_read(&rioc->rdmain);
3261
3262	if (!rdma) {
3263	return -1;
3264	}
3265
3266	if (rdma_errored(rdma)) {
3267	return -1;
3268	}
3269
3270	local = &rdma->local_ram_blocks;
3271	do {
3272	trace_rdma_registration_handle_wait();
3273
3274	ret = qemu_rdma_exchange_recv(rdma, &head, RDMA_CONTROL_NONE, &err);
3275
3276	if (ret < 0) {
3277	error_report_err(err);
3278	break;
3279	}
3280
3281	if (head.repeat > RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE) {
3282	error_report("rdma: Too many requests in this message (%d)."
3283	"Bailing.", head.repeat);
3284	break;
3285	}
3286
3287	switch (head.type) {
3288	case RDMA_CONTROL_COMPRESS:
3289	comp = (RDMACompress *) rdma->wr_data[idx].control_curr;
3290	network_to_compress(comp);
3291
3292	trace_rdma_registration_handle_compress(comp->length,
3293	comp->block_idx,
3294	comp->offset);
3295	if (comp->block_idx >= rdma->local_ram_blocks.nb_blocks) {
3296	error_report("rdma: 'compress' bad block index %u (vs %d)",
3297	(unsigned int)comp->block_idx,
3298	rdma->local_ram_blocks.nb_blocks);
3299	goto err;
3300	}
3301	block = &(rdma->local_ram_blocks.block[comp->block_idx]);
3302	if (!rdma_compress_range_check(block, comp)) {
3303	goto err;
3304	}
3305	host_addr = block->local_host_addr +
3306	(comp->offset - block->offset);
3307	if (comp->value) {
3308	error_report("rdma: Zero page with non-zero (%d) value",
3309	comp->value);
3310	goto err;
3311	}
3312	ram_handle_zero(host_addr, comp->length);
3313	break;
3314
3315	case RDMA_CONTROL_REGISTER_FINISHED:
3316	trace_rdma_registration_handle_finished();
3317	return 0;
3318
3319	case RDMA_CONTROL_RAM_BLOCKS_REQUEST:
3320	trace_rdma_registration_handle_ram_blocks();
3321
3322	/* Sort our local RAM Block list so it's the same as the source,
3323	* we can do this since we've filled in a src_index in the list
3324	* as we received the RAMBlock list earlier.
3325	*/
3326	qsort(rdma->local_ram_blocks.block,
3327	rdma->local_ram_blocks.nb_blocks,
3328	sizeof(RDMALocalBlock), dest_ram_sort_func);
3329	for (int i = 0; i < local->nb_blocks; i++) {
3330	local->block[i].index = i;
3331	}
3332
3333	if (rdma->pin_all) {
3334	ret = qemu_rdma_reg_whole_ram_blocks(rdma, &err);
3335	if (ret < 0) {
3336	error_report_err(err);
3337	goto err;
3338	}
3339	}
3340
3341	/*
3342	* Dest uses this to prepare to transmit the RAMBlock descriptions
3343	* to the source VM after connection setup.
3344	* Both sides use the "remote" structure to communicate and update
3345	* their "local" descriptions with what was sent.
3346	*/
3347	for (int i = 0; i < local->nb_blocks; i++) {
3348	rdma->dest_blocks[i].remote_host_addr =
3349	(uintptr_t)(local->block[i].local_host_addr);
3350
3351	if (rdma->pin_all) {
3352	rdma->dest_blocks[i].remote_rkey = local->block[i].mr->rkey;
3353	}
3354
3355	rdma->dest_blocks[i].offset = local->block[i].offset;
3356	rdma->dest_blocks[i].length = local->block[i].length;
3357
3358	dest_block_to_network(&rdma->dest_blocks[i]);
3359	trace_rdma_registration_handle_ram_blocks_loop(
3360	local->block[i].block_name,
3361	local->block[i].offset,
3362	local->block[i].length,
3363	local->block[i].local_host_addr,
3364	local->block[i].src_index);
3365	}
3366
3367	blocks.len = rdma->local_ram_blocks.nb_blocks
3368	* sizeof(RDMADestBlock);
3369
3370
3371	ret = qemu_rdma_post_send_control(rdma,
3372	(uint8_t *) rdma->dest_blocks, &blocks,
3373	&err);
3374
3375	if (ret < 0) {
3376	error_report_err(err);
3377	goto err;
3378	}
3379
3380	break;
3381	case RDMA_CONTROL_REGISTER_REQUEST:
3382	trace_rdma_registration_handle_register(head.repeat);
3383
3384	reg_resp.repeat = head.repeat;
3385	registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
3386
3387	/* Making sure the register buffers to read are valid */
3388	if (head.len != head.repeat * sizeof(RDMARegister)) {
3389	error_report("%s: Invalid RDMA_CONTROL_REGISTER_REQUEST "
3390	"(head.repeat=%"PRIu32", head.len=%"PRIu32")",
3391	__func__, head.repeat, head.len);
3392	goto err;
3393	}
3394
3395	for (int count = 0; count < head.repeat; count++) {
3396	uint64_t chunk, chunk_sum;
3397	uint8_t chunk_start, chunk_end;
3398
3399	reg = &registers[count];
3400	network_to_register(reg);
3401
3402	reg_result = &results[count];
3403
3404	trace_rdma_registration_handle_register_loop(count,
3405	reg->current_index, reg->current_addr, reg->chunks);
3406
3407	if (reg->current_index >= rdma->local_ram_blocks.nb_blocks) {
3408	error_report("rdma: 'register' bad block index %u (vs %d)",
3409	(unsigned int)reg->current_index,
3410	rdma->local_ram_blocks.nb_blocks);
3411	goto err;
3412	}
3413	block = &(rdma->local_ram_blocks.block[reg->current_index]);
3414	if (block->offset > reg->current_addr \|\|
3415	block->offset + block->length <= reg->current_addr) {
3416	error_report("rdma: bad register address for block %s"
3417	" offset: %" PRIx64 " current_addr: %" PRIx64,
3418	block->block_name, block->offset,
3419	reg->current_addr);
3420	goto err;
3421	}
3422	host_addr = (block->local_host_addr +
3423	(reg->current_addr - block->offset));
3424	chunk = ram_chunk_index(block->local_host_addr,
3425	(uint8_t *) host_addr);
3426	chunk_start = ram_chunk_start(block, chunk);
3427	if (uadd64_overflow(chunk, reg->chunks, &chunk_sum) \|\|
3428	chunk_sum >= block->nb_chunks) {
3429	error_report("%s: head.chunks contains illegal value"
3430	" (chunk=%"PRIu64", chunks=%"PRIu64", "
3431	"nb_chunks=%d)", __func__, chunk,
3432	reg->chunks, block->nb_chunks);
3433	goto err;
3434	}
3435	chunk_end = ram_chunk_end(block, chunk + reg->chunks);
3436	/* avoid "-Waddress-of-packed-member" warning */
3437	uint32_t tmp_rkey = 0;
3438	if (qemu_rdma_register_and_get_keys(rdma, block,
3439	(uintptr_t)host_addr, NULL, &tmp_rkey,
3440	chunk, chunk_start, chunk_end)) {
3441	error_report("cannot get rkey");
3442	goto err;
3443	}
3444	reg_result->rkey = tmp_rkey;
3445
3446	reg_result->host_addr = (uintptr_t)block->local_host_addr;
3447
3448	trace_rdma_registration_handle_register_rkey(reg_result->rkey);
3449
3450	result_to_network(reg_result);
3451	}
3452
3453	ret = qemu_rdma_post_send_control(rdma,
3454	(uint8_t *) results, &reg_resp, &err);
3455
3456	if (ret < 0) {
3457	error_report_err(err);
3458	goto err;
3459	}
3460	break;
3461	case RDMA_CONTROL_REGISTER_RESULT:
3462	error_report("Invalid RESULT message at dest.");
3463	goto err;
3464	default:
3465	error_report("Unknown control message %s", control_desc(head.type));
3466	goto err;
3467	}
3468	} while (1);
3469
3470	err:
3471	rdma->errored = true;
3472	return -1;
3473	}
3474
3475	/* Destination:
3476	* Called during the initial RAM load section which lists the
3477	* RAMBlocks by name. This lets us know the order of the RAMBlocks on
3478	* the source. We've already built our local RAMBlock list, but not
3479	* yet sent the list to the source.
3480	*/
3481	int rdma_block_notification_handle(QEMUFile f, const char name)
3482	{
3483	int curr;
3484	int found = -1;
3485
3486	if (!migrate_rdma()) {
3487	return 0;
3488	}
3489
3490	RCU_READ_LOCK_GUARD();
3491	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3492	RDMAContext *rdma = qatomic_rcu_read(&rioc->rdmain);
3493
3494	if (!rdma) {
3495	return -1;
3496	}
3497
3498	/* Find the matching RAMBlock in our local list */
3499	for (curr = 0; curr < rdma->local_ram_blocks.nb_blocks; curr++) {
3500	if (!strcmp(rdma->local_ram_blocks.block[curr].block_name, name)) {
3501	found = curr;
3502	break;
3503	}
3504	}
3505
3506	if (found == -1) {
3507	error_report("RAMBlock '%s' not found on destination", name);
3508	return -1;
3509	}
3510
3511	rdma->local_ram_blocks.block[curr].src_index = rdma->next_src_index;
3512	trace_rdma_block_notification_handle(name, rdma->next_src_index);
3513	rdma->next_src_index++;
3514
3515	return 0;
3516	}
3517
3518	int rdma_registration_start(QEMUFile *f, uint64_t flags)
3519	{
3520	if (!migrate_rdma()) {
3521	return 0;
3522	}
3523
3524	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3525	RCU_READ_LOCK_GUARD();
3526	RDMAContext *rdma = qatomic_rcu_read(&rioc->rdmaout);
3527	if (!rdma) {
3528	return -1;
3529	}
3530
3531	if (rdma_errored(rdma)) {
3532	return -1;
3533	}
3534
3535	trace_rdma_registration_start(flags);
3536	qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
3537	return qemu_fflush(f);
3538	}
3539
3540	/*
3541	* Inform dest that dynamic registrations are done for now.
3542	* First, flush writes, if any.
3543	*/
3544	int rdma_registration_stop(QEMUFile *f, uint64_t flags)
3545	{
3546	QIOChannelRDMA *rioc;
3547	Error *err = NULL;
3548	RDMAContext *rdma;
3549	RDMAControlHeader head = { .len = 0, .repeat = 1 };
3550	int ret;
3551
3552	if (!migrate_rdma()) {
3553	return 0;
3554	}
3555
3556	RCU_READ_LOCK_GUARD();
3557	rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
3558	rdma = qatomic_rcu_read(&rioc->rdmaout);
3559	if (!rdma) {
3560	return -1;
3561	}
3562
3563	if (rdma_errored(rdma)) {
3564	return -1;
3565	}
3566
3567	qemu_fflush(f);
3568	ret = qemu_rdma_drain_cq(rdma);
3569
3570	if (ret < 0) {
3571	goto err;
3572	}
3573
3574	if (flags == RAM_CONTROL_SETUP) {
3575	RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
3576	RDMALocalBlocks *local = &rdma->local_ram_blocks;
3577	int reg_result_idx, nb_dest_blocks;
3578
3579	head.type = RDMA_CONTROL_RAM_BLOCKS_REQUEST;
3580	trace_rdma_registration_stop_ram();
3581
3582	/*
3583	* Make sure that we parallelize the pinning on both sides.
3584	* For very large guests, doing this serially takes a really
3585	* long time, so we have to 'interleave' the pinning locally
3586	* with the control messages by performing the pinning on this
3587	* side before we receive the control response from the other
3588	* side that the pinning has completed.
3589	*/
3590	ret = qemu_rdma_exchange_send(rdma, &head, NULL, &resp,
3591	&reg_result_idx, rdma->pin_all ?
3592	qemu_rdma_reg_whole_ram_blocks : NULL,
3593	&err);
3594	if (ret < 0) {
3595	error_report_err(err);
3596	return -1;
3597	}
3598
3599	nb_dest_blocks = resp.len / sizeof(RDMADestBlock);
3600
3601	/*
3602	* The protocol uses two different sets of rkeys (mutually exclusive):
3603	* 1. One key to represent the virtual address of the entire ram block.
3604	* (dynamic chunk registration disabled - pin everything with one rkey.)
3605	* 2. One to represent individual chunks within a ram block.
3606	* (dynamic chunk registration enabled - pin individual chunks.)
3607	*
3608	* Once the capability is successfully negotiated, the destination transmits
3609	* the keys to use (or sends them later) including the virtual addresses
3610	* and then propagates the remote ram block descriptions to his local copy.
3611	*/
3612
3613	if (local->nb_blocks != nb_dest_blocks) {
3614	error_report("ram blocks mismatch (Number of blocks %d vs %d)",
3615	local->nb_blocks, nb_dest_blocks);
3616	error_printf("Your QEMU command line parameters are probably "
3617	"not identical on both the source and destination.");
3618	rdma->errored = true;
3619	return -1;
3620	}
3621
3622	qemu_rdma_move_header(rdma, reg_result_idx, &resp);
3623	memcpy(rdma->dest_blocks,
3624	rdma->wr_data[reg_result_idx].control_curr, resp.len);
3625	for (int i = 0; i < nb_dest_blocks; i++) {
3626	network_to_dest_block(&rdma->dest_blocks[i]);
3627
3628	/* We require that the blocks are in the same order */
3629	if (rdma->dest_blocks[i].length != local->block[i].length) {
3630	error_report("Block %s/%d has a different length %" PRIu64
3631	"vs %" PRIu64,
3632	local->block[i].block_name, i,
3633	local->block[i].length,
3634	rdma->dest_blocks[i].length);
3635	rdma->errored = true;
3636	return -1;
3637	}
3638	local->block[i].remote_host_addr =
3639	rdma->dest_blocks[i].remote_host_addr;
3640	local->block[i].remote_rkey = rdma->dest_blocks[i].remote_rkey;
3641	}
3642	}
3643
3644	trace_rdma_registration_stop(flags);
3645
3646	head.type = RDMA_CONTROL_REGISTER_FINISHED;
3647	ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL, &err);
3648
3649	if (ret < 0) {
3650	error_report_err(err);
3651	goto err;
3652	}
3653
3654	return 0;
3655	err:
3656	rdma->errored = true;
3657	return -1;
3658	}
3659
3660	static void qio_channel_rdma_finalize(Object *obj)
3661	{
3662	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(obj);
3663	if (rioc->rdmain) {
3664	qemu_rdma_cleanup(rioc->rdmain);
3665	g_free(rioc->rdmain);
3666	rioc->rdmain = NULL;
3667	}
3668	if (rioc->rdmaout) {
3669	qemu_rdma_cleanup(rioc->rdmaout);
3670	g_free(rioc->rdmaout);
3671	rioc->rdmaout = NULL;
3672	}
3673	}
3674
3675	static void qio_channel_rdma_class_init(ObjectClass *klass,
3676	const void *class_data G_GNUC_UNUSED)
3677	{
3678	QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
3679
3680	ioc_klass->io_writev = qio_channel_rdma_writev;
3681	ioc_klass->io_readv = qio_channel_rdma_readv;
3682	ioc_klass->io_set_blocking = qio_channel_rdma_set_blocking;
3683	ioc_klass->io_close = qio_channel_rdma_close;
3684	ioc_klass->io_create_watch = qio_channel_rdma_create_watch;
3685	ioc_klass->io_set_aio_fd_handler = qio_channel_rdma_set_aio_fd_handler;
3686	ioc_klass->io_shutdown = qio_channel_rdma_shutdown;
3687	}
3688
3689	static const TypeInfo qio_channel_rdma_info = {
3690	.parent = TYPE_QIO_CHANNEL,
3691	.name = TYPE_QIO_CHANNEL_RDMA,
3692	.instance_size = sizeof(QIOChannelRDMA),
3693	.instance_finalize = qio_channel_rdma_finalize,
3694	.class_init = qio_channel_rdma_class_init,
3695	};
3696
3697	static void qio_channel_rdma_register_types(void)
3698	{
3699	type_register_static(&qio_channel_rdma_info);
3700	}
3701
3702	type_init(qio_channel_rdma_register_types);
3703
3704	static QIOChannel rdma_new_input(RDMAContext rdma)
3705	{
3706	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA));
3707
3708	rioc->rdmain = rdma;
3709	rioc->rdmaout = rdma->return_path;
3710
3711	return QIO_CHANNEL(rioc);
3712	}
3713
3714	static QIOChannel rdma_new_output(RDMAContext rdma)
3715	{
3716	QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(object_new(TYPE_QIO_CHANNEL_RDMA));
3717
3718	rioc->rdmaout = rdma;
3719	rioc->rdmain = rdma->return_path;
3720
3721	return QIO_CHANNEL(rioc);
3722	}
3723
3724	static void rdma_accept_incoming_migration(void *opaque)
3725	{
3726	RDMAContext *rdma = opaque;
3727	QIOChannel *ioc;
3728
3729	trace_rdma_accept_incoming_migration();
3730	if (qemu_rdma_accept(rdma) < 0) {
3731	error_report("RDMA ERROR: Migration initialization failed");
3732	return;
3733	}
3734
3735	trace_rdma_accept_incoming_migration_accepted();
3736
3737	if (rdma->is_return_path) {
3738	return;
3739	}
3740
3741	ioc = rdma_new_input(rdma);
3742	if (ioc == NULL) {
3743	error_report("RDMA ERROR: could not open RDMA for input");
3744	qemu_rdma_cleanup(rdma);
3745	return;
3746	}
3747
3748	rdma->migration_started_on_destination = 1;
3749	migration_incoming_setup(ioc, CH_MAIN, &error_abort);
3750	migration_start_incoming();
3751	}
3752
3753	void rdma_connect_incoming(InetSocketAddress host_port, Error *errp)
3754	{
3755	MigrationState *s = migrate_get_current();
3756	int ret;
3757	RDMAContext *rdma;
3758
3759	trace_rdma_connect_incoming();
3760
3761	/* Avoid ram_block_discard_disable(), cannot change during migration. */
3762	if (ram_block_discard_is_required()) {
3763	error_setg(errp, "RDMA: cannot disable RAM discard");
3764	return;
3765	}
3766
3767	rdma = qemu_rdma_data_init(host_port, errp);
3768	if (rdma == NULL) {
3769	goto err;
3770	}
3771
3772	ret = qemu_rdma_dest_init(rdma, errp);
3773	if (ret < 0) {
3774	goto err;
3775	}
3776
3777	trace_rdma_connect_incoming_after_dest_init();
3778
3779	ret = rdma_listen(rdma->listen_id, 5);
3780
3781	if (ret < 0) {
3782	error_setg(errp, "RDMA ERROR: listening on socket!");
3783	goto cleanup_rdma;
3784	}
3785
3786	trace_rdma_connect_incoming_after_rdma_listen();
3787	s->rdma_migration = true;
3788	qemu_set_fd_handler(rdma->channel->fd, rdma_accept_incoming_migration,
3789	NULL, (void *)(intptr_t)rdma);
3790	return;
3791
3792	cleanup_rdma:
3793	qemu_rdma_cleanup(rdma);
3794	err:
3795	if (rdma) {
3796	g_free(rdma->host);
3797	}
3798	g_free(rdma);
3799	}
3800
3801	QIOChannel rdma_connect_outgoing(void opaque,
3802	InetSocketAddress host_port, Error *errp)
3803	{
3804	MigrationState *s = opaque;
3805	RDMAContext *rdma_return_path = NULL;
3806	RDMAContext *rdma;
3807	int ret;
3808
3809	/* Avoid ram_block_discard_disable(), cannot change during migration. */
3810	if (ram_block_discard_is_required()) {
3811	error_setg(errp, "RDMA: cannot disable RAM discard");
3812	return NULL;
3813	}
3814
3815	rdma = qemu_rdma_data_init(host_port, errp);
3816	if (rdma == NULL) {
3817	goto err;
3818	}
3819
3820	ret = qemu_rdma_source_init(rdma, migrate_rdma_pin_all(), errp);
3821
3822	if (ret < 0) {
3823	goto err;
3824	}
3825
3826	trace_rdma_connect_outgoing_after_rdma_source_init();
3827	ret = qemu_rdma_connect(rdma, false, errp);
3828
3829	if (ret < 0) {
3830	goto err;
3831	}
3832
3833	/* RDMA postcopy need a separate queue pair for return path */
3834	if (migrate_postcopy() \|\| migrate_return_path()) {
3835	rdma_return_path = qemu_rdma_data_init(host_port, errp);
3836
3837	if (rdma_return_path == NULL) {
3838	goto return_path_err;
3839	}
3840
3841	ret = qemu_rdma_source_init(rdma_return_path,
3842	migrate_rdma_pin_all(), errp);
3843
3844	if (ret < 0) {
3845	goto return_path_err;
3846	}
3847
3848	ret = qemu_rdma_connect(rdma_return_path, true, errp);
3849
3850	if (ret < 0) {
3851	goto return_path_err;
3852	}
3853
3854	rdma->return_path = rdma_return_path;
3855	rdma_return_path->return_path = rdma;
3856	rdma_return_path->is_return_path = true;
3857	}
3858
3859	trace_rdma_connect_outgoing_after_rdma_connect();
3860
3861	s->rdma_migration = true;
3862	return rdma_new_output(rdma);
3863	return_path_err:
3864	qemu_rdma_cleanup(rdma);
3865	err:
3866	g_free(rdma);
3867	g_free(rdma_return_path);
3868	return NULL;
3869	}