1 // SPDX-License-Identifier: GPL-2.0-only
2 /*
3  * VMware VMCI Driver
4  *
5  * Copyright (C) 2012 VMware, Inc. All rights reserved.
6  */
7 
8 #include <linux/vmw_vmci_defs.h>
9 #include <linux/vmw_vmci_api.h>
10 #include <linux/highmem.h>
11 #include <linux/kernel.h>
12 #include <linux/mm.h>
13 #include <linux/module.h>
14 #include <linux/mutex.h>
15 #include <linux/pagemap.h>
16 #include <linux/pci.h>
17 #include <linux/sched.h>
18 #include <linux/slab.h>
19 #include <linux/uio.h>
20 #include <linux/wait.h>
21 #include <linux/vmalloc.h>
22 #include <linux/skbuff.h>
23 
24 #include "vmci_handle_array.h"
25 #include "vmci_queue_pair.h"
26 #include "vmci_datagram.h"
27 #include "vmci_resource.h"
28 #include "vmci_context.h"
29 #include "vmci_driver.h"
30 #include "vmci_event.h"
31 #include "vmci_route.h"
32 
33 /*
34  * In the following, we will distinguish between two kinds of VMX processes -
35  * the ones with versions lower than VMCI_VERSION_NOVMVM that use specialized
36  * VMCI page files in the VMX and supporting VM to VM communication and the
37  * newer ones that use the guest memory directly. We will in the following
38  * refer to the older VMX versions as old-style VMX'en, and the newer ones as
39  * new-style VMX'en.
40  *
41  * The state transition datagram is as follows (the VMCIQPB_ prefix has been
42  * removed for readability) - see below for more details on the transtions:
43  *
44  *            --------------  NEW  -------------
45  *            |                                |
46  *           \_/                              \_/
47  *     CREATED_NO_MEM <-----------------> CREATED_MEM
48  *            |    |                           |
49  *            |    o-----------------------o   |
50  *            |                            |   |
51  *           \_/                          \_/ \_/
52  *     ATTACHED_NO_MEM <----------------> ATTACHED_MEM
53  *            |                            |   |
54  *            |     o----------------------o   |
55  *            |     |                          |
56  *           \_/   \_/                        \_/
57  *     SHUTDOWN_NO_MEM <----------------> SHUTDOWN_MEM
58  *            |                                |
59  *            |                                |
60  *            -------------> gone <-------------
61  *
62  * In more detail. When a VMCI queue pair is first created, it will be in the
63  * VMCIQPB_NEW state. It will then move into one of the following states:
64  *
65  * - VMCIQPB_CREATED_NO_MEM: this state indicates that either:
66  *
67  *     - the created was performed by a host endpoint, in which case there is
68  *       no backing memory yet.
69  *
70  *     - the create was initiated by an old-style VMX, that uses
71  *       vmci_qp_broker_set_page_store to specify the UVAs of the queue pair at
72  *       a later point in time. This state can be distinguished from the one
73  *       above by the context ID of the creator. A host side is not allowed to
74  *       attach until the page store has been set.
75  *
76  * - VMCIQPB_CREATED_MEM: this state is the result when the queue pair
77  *     is created by a VMX using the queue pair device backend that
78  *     sets the UVAs of the queue pair immediately and stores the
79  *     information for later attachers. At this point, it is ready for
80  *     the host side to attach to it.
81  *
82  * Once the queue pair is in one of the created states (with the exception of
83  * the case mentioned for older VMX'en above), it is possible to attach to the
84  * queue pair. Again we have two new states possible:
85  *
86  * - VMCIQPB_ATTACHED_MEM: this state can be reached through the following
87  *   paths:
88  *
89  *     - from VMCIQPB_CREATED_NO_MEM when a new-style VMX allocates a queue
90  *       pair, and attaches to a queue pair previously created by the host side.
91  *
92  *     - from VMCIQPB_CREATED_MEM when the host side attaches to a queue pair
93  *       already created by a guest.
94  *
95  *     - from VMCIQPB_ATTACHED_NO_MEM, when an old-style VMX calls
96  *       vmci_qp_broker_set_page_store (see below).
97  *
98  * - VMCIQPB_ATTACHED_NO_MEM: If the queue pair already was in the
99  *     VMCIQPB_CREATED_NO_MEM due to a host side create, an old-style VMX will
100  *     bring the queue pair into this state. Once vmci_qp_broker_set_page_store
101  *     is called to register the user memory, the VMCIQPB_ATTACH_MEM state
102  *     will be entered.
103  *
104  * From the attached queue pair, the queue pair can enter the shutdown states
105  * when either side of the queue pair detaches. If the guest side detaches
106  * first, the queue pair will enter the VMCIQPB_SHUTDOWN_NO_MEM state, where
107  * the content of the queue pair will no longer be available. If the host
108  * side detaches first, the queue pair will either enter the
109  * VMCIQPB_SHUTDOWN_MEM, if the guest memory is currently mapped, or
110  * VMCIQPB_SHUTDOWN_NO_MEM, if the guest memory is not mapped
111  * (e.g., the host detaches while a guest is stunned).
112  *
113  * New-style VMX'en will also unmap guest memory, if the guest is
114  * quiesced, e.g., during a snapshot operation. In that case, the guest
115  * memory will no longer be available, and the queue pair will transition from
116  * *_MEM state to a *_NO_MEM state. The VMX may later map the memory once more,
117  * in which case the queue pair will transition from the *_NO_MEM state at that
118  * point back to the *_MEM state. Note that the *_NO_MEM state may have changed,
119  * since the peer may have either attached or detached in the meantime. The
120  * values are laid out such that ++ on a state will move from a *_NO_MEM to a
121  * *_MEM state, and vice versa.
122  */
123 
124 /* The Kernel specific component of the struct vmci_queue structure. */
125 struct vmci_queue_kern_if {
126 	struct mutex __mutex;	/* Protects the queue. */
127 	struct mutex *mutex;	/* Shared by producer and consumer queues. */
128 	size_t num_pages;	/* Number of pages incl. header. */
129 	bool host;		/* Host or guest? */
130 	union {
131 		struct {
132 			dma_addr_t *pas;
133 			void **vas;
134 		} g;		/* Used by the guest. */
135 		struct {
136 			struct page **page;
137 			struct page **header_page;
138 		} h;		/* Used by the host. */
139 	} u;
140 };
141 
142 /*
143  * This structure is opaque to the clients.
144  */
145 struct vmci_qp {
146 	struct vmci_handle handle;
147 	struct vmci_queue *produce_q;
148 	struct vmci_queue *consume_q;
149 	u64 produce_q_size;
150 	u64 consume_q_size;
151 	u32 peer;
152 	u32 flags;
153 	u32 priv_flags;
154 	bool guest_endpoint;
155 	unsigned int blocked;
156 	unsigned int generation;
157 	wait_queue_head_t event;
158 };
159 
160 enum qp_broker_state {
161 	VMCIQPB_NEW,
162 	VMCIQPB_CREATED_NO_MEM,
163 	VMCIQPB_CREATED_MEM,
164 	VMCIQPB_ATTACHED_NO_MEM,
165 	VMCIQPB_ATTACHED_MEM,
166 	VMCIQPB_SHUTDOWN_NO_MEM,
167 	VMCIQPB_SHUTDOWN_MEM,
168 	VMCIQPB_GONE
169 };
170 
171 #define QPBROKERSTATE_HAS_MEM(_qpb) (_qpb->state == VMCIQPB_CREATED_MEM || \
172 				     _qpb->state == VMCIQPB_ATTACHED_MEM || \
173 				     _qpb->state == VMCIQPB_SHUTDOWN_MEM)
174 
175 /*
176  * In the queue pair broker, we always use the guest point of view for
177  * the produce and consume queue values and references, e.g., the
178  * produce queue size stored is the guests produce queue size. The
179  * host endpoint will need to swap these around. The only exception is
180  * the local queue pairs on the host, in which case the host endpoint
181  * that creates the queue pair will have the right orientation, and
182  * the attaching host endpoint will need to swap.
183  */
184 struct qp_entry {
185 	struct list_head list_item;
186 	struct vmci_handle handle;
187 	u32 peer;
188 	u32 flags;
189 	u64 produce_size;
190 	u64 consume_size;
191 	u32 ref_count;
192 };
193 
194 struct qp_broker_entry {
195 	struct vmci_resource resource;
196 	struct qp_entry qp;
197 	u32 create_id;
198 	u32 attach_id;
199 	enum qp_broker_state state;
200 	bool require_trusted_attach;
201 	bool created_by_trusted;
202 	bool vmci_page_files;	/* Created by VMX using VMCI page files */
203 	struct vmci_queue *produce_q;
204 	struct vmci_queue *consume_q;
205 	struct vmci_queue_header saved_produce_q;
206 	struct vmci_queue_header saved_consume_q;
207 	vmci_event_release_cb wakeup_cb;
208 	void *client_data;
209 	void *local_mem;	/* Kernel memory for local queue pair */
210 };
211 
212 struct qp_guest_endpoint {
213 	struct vmci_resource resource;
214 	struct qp_entry qp;
215 	u64 num_ppns;
216 	void *produce_q;
217 	void *consume_q;
218 	struct ppn_set ppn_set;
219 };
220 
221 struct qp_list {
222 	struct list_head head;
223 	struct mutex mutex;	/* Protect queue list. */
224 };
225 
226 static struct qp_list qp_broker_list = {
227 	.head = LIST_HEAD_INIT(qp_broker_list.head),
228 	.mutex = __MUTEX_INITIALIZER(qp_broker_list.mutex),
229 };
230 
231 static struct qp_list qp_guest_endpoints = {
232 	.head = LIST_HEAD_INIT(qp_guest_endpoints.head),
233 	.mutex = __MUTEX_INITIALIZER(qp_guest_endpoints.mutex),
234 };
235 
236 #define INVALID_VMCI_GUEST_MEM_ID  0
237 #define QPE_NUM_PAGES(_QPE) ((u32) \
238 			     (DIV_ROUND_UP(_QPE.produce_size, PAGE_SIZE) + \
239 			      DIV_ROUND_UP(_QPE.consume_size, PAGE_SIZE) + 2))
240 #define QP_SIZES_ARE_VALID(_prod_qsize, _cons_qsize) \
241 	((_prod_qsize) + (_cons_qsize) >= max(_prod_qsize, _cons_qsize) && \
242 	 (_prod_qsize) + (_cons_qsize) <= VMCI_MAX_GUEST_QP_MEMORY)
243 
244 /*
245  * Frees kernel VA space for a given queue and its queue header, and
246  * frees physical data pages.
247  */
248 static void qp_free_queue(void *q, u64 size)
249 {
250 	struct vmci_queue *queue = q;
251 
252 	if (queue) {
253 		u64 i;
254 
255 		/* Given size does not include header, so add in a page here. */
256 		for (i = 0; i < DIV_ROUND_UP(size, PAGE_SIZE) + 1; i++) {
257 			dma_free_coherent(&vmci_pdev->dev, PAGE_SIZE,
258 					  queue->kernel_if->u.g.vas[i],
259 					  queue->kernel_if->u.g.pas[i]);
260 		}
261 
262 		vfree(queue);
263 	}
264 }
265 
266 /*
267  * Allocates kernel queue pages of specified size with IOMMU mappings,
268  * plus space for the queue structure/kernel interface and the queue
269  * header.
270  */
271 static void *qp_alloc_queue(u64 size, u32 flags)
272 {
273 	u64 i;
274 	struct vmci_queue *queue;
275 	size_t pas_size;
276 	size_t vas_size;
277 	size_t queue_size = sizeof(*queue) + sizeof(*queue->kernel_if);
278 	u64 num_pages;
279 
280 	if (size > SIZE_MAX - PAGE_SIZE)
281 		return NULL;
282 	num_pages = DIV_ROUND_UP(size, PAGE_SIZE) + 1;
283 	if (num_pages >
284 		 (SIZE_MAX - queue_size) /
285 		 (sizeof(*queue->kernel_if->u.g.pas) +
286 		  sizeof(*queue->kernel_if->u.g.vas)))
287 		return NULL;
288 
289 	pas_size = num_pages * sizeof(*queue->kernel_if->u.g.pas);
290 	vas_size = num_pages * sizeof(*queue->kernel_if->u.g.vas);
291 	queue_size += pas_size + vas_size;
292 
293 	queue = vmalloc(queue_size);
294 	if (!queue)
295 		return NULL;
296 
297 	queue->q_header = NULL;
298 	queue->saved_header = NULL;
299 	queue->kernel_if = (struct vmci_queue_kern_if *)(queue + 1);
300 	queue->kernel_if->mutex = NULL;
301 	queue->kernel_if->num_pages = num_pages;
302 	queue->kernel_if->u.g.pas = (dma_addr_t *)(queue->kernel_if + 1);
303 	queue->kernel_if->u.g.vas =
304 		(void **)((u8 *)queue->kernel_if->u.g.pas + pas_size);
305 	queue->kernel_if->host = false;
306 
307 	for (i = 0; i < num_pages; i++) {
308 		queue->kernel_if->u.g.vas[i] =
309 			dma_alloc_coherent(&vmci_pdev->dev, PAGE_SIZE,
310 					   &queue->kernel_if->u.g.pas[i],
311 					   GFP_KERNEL);
312 		if (!queue->kernel_if->u.g.vas[i]) {
313 			/* Size excl. the header. */
314 			qp_free_queue(queue, i * PAGE_SIZE);
315 			return NULL;
316 		}
317 	}
318 
319 	/* Queue header is the first page. */
320 	queue->q_header = queue->kernel_if->u.g.vas[0];
321 
322 	return queue;
323 }
324 
325 /*
326  * Copies from a given buffer or iovector to a VMCI Queue.  Uses
327  * kmap_local_page() to dynamically map required portions of the queue
328  * by traversing the offset -> page translation structure for the queue.
329  * Assumes that offset + size does not wrap around in the queue.
330  */
331 static int qp_memcpy_to_queue_iter(struct vmci_queue *queue,
332 				  u64 queue_offset,
333 				  struct iov_iter *from,
334 				  size_t size)
335 {
336 	struct vmci_queue_kern_if *kernel_if = queue->kernel_if;
337 	size_t bytes_copied = 0;
338 
339 	while (bytes_copied < size) {
340 		const u64 page_index =
341 			(queue_offset + bytes_copied) / PAGE_SIZE;
342 		const size_t page_offset =
343 		    (queue_offset + bytes_copied) & (PAGE_SIZE - 1);
344 		void *va;
345 		size_t to_copy;
346 
347 		if (kernel_if->host)
348 			va = kmap_local_page(kernel_if->u.h.page[page_index]);
349 		else
350 			va = kernel_if->u.g.vas[page_index + 1];
351 			/* Skip header. */
352 
353 		if (size - bytes_copied > PAGE_SIZE - page_offset)
354 			/* Enough payload to fill up from this page. */
355 			to_copy = PAGE_SIZE - page_offset;
356 		else
357 			to_copy = size - bytes_copied;
358 
359 		if (!copy_from_iter_full((u8 *)va + page_offset, to_copy,
360 					 from)) {
361 			if (kernel_if->host)
362 				kunmap_local(va);
363 			return VMCI_ERROR_INVALID_ARGS;
364 		}
365 		bytes_copied += to_copy;
366 		if (kernel_if->host)
367 			kunmap_local(va);
368 	}
369 
370 	return VMCI_SUCCESS;
371 }
372 
373 /*
374  * Copies to a given buffer or iovector from a VMCI Queue.  Uses
375  * kmap_local_page() to dynamically map required portions of the queue
376  * by traversing the offset -> page translation structure for the queue.
377  * Assumes that offset + size does not wrap around in the queue.
378  */
379 static int qp_memcpy_from_queue_iter(struct iov_iter *to,
380 				    const struct vmci_queue *queue,
381 				    u64 queue_offset, size_t size)
382 {
383 	struct vmci_queue_kern_if *kernel_if = queue->kernel_if;
384 	size_t bytes_copied = 0;
385 
386 	while (bytes_copied < size) {
387 		const u64 page_index =
388 			(queue_offset + bytes_copied) / PAGE_SIZE;
389 		const size_t page_offset =
390 		    (queue_offset + bytes_copied) & (PAGE_SIZE - 1);
391 		void *va;
392 		size_t to_copy;
393 		int err;
394 
395 		if (kernel_if->host)
396 			va = kmap_local_page(kernel_if->u.h.page[page_index]);
397 		else
398 			va = kernel_if->u.g.vas[page_index + 1];
399 			/* Skip header. */
400 
401 		if (size - bytes_copied > PAGE_SIZE - page_offset)
402 			/* Enough payload to fill up this page. */
403 			to_copy = PAGE_SIZE - page_offset;
404 		else
405 			to_copy = size - bytes_copied;
406 
407 		err = copy_to_iter((u8 *)va + page_offset, to_copy, to);
408 		if (err != to_copy) {
409 			if (kernel_if->host)
410 				kunmap_local(va);
411 			return VMCI_ERROR_INVALID_ARGS;
412 		}
413 		bytes_copied += to_copy;
414 		if (kernel_if->host)
415 			kunmap_local(va);
416 	}
417 
418 	return VMCI_SUCCESS;
419 }
420 
421 /*
422  * Allocates two list of PPNs --- one for the pages in the produce queue,
423  * and the other for the pages in the consume queue. Intializes the list
424  * of PPNs with the page frame numbers of the KVA for the two queues (and
425  * the queue headers).
426  */
427 static int qp_alloc_ppn_set(void *prod_q,
428 			    u64 num_produce_pages,
429 			    void *cons_q,
430 			    u64 num_consume_pages, struct ppn_set *ppn_set)
431 {
432 	u64 *produce_ppns;
433 	u64 *consume_ppns;
434 	struct vmci_queue *produce_q = prod_q;
435 	struct vmci_queue *consume_q = cons_q;
436 	u64 i;
437 
438 	if (!produce_q || !num_produce_pages || !consume_q ||
439 	    !num_consume_pages || !ppn_set)
440 		return VMCI_ERROR_INVALID_ARGS;
441 
442 	if (ppn_set->initialized)
443 		return VMCI_ERROR_ALREADY_EXISTS;
444 
445 	produce_ppns =
446 	    kmalloc_array(num_produce_pages, sizeof(*produce_ppns),
447 			  GFP_KERNEL);
448 	if (!produce_ppns)
449 		return VMCI_ERROR_NO_MEM;
450 
451 	consume_ppns =
452 	    kmalloc_array(num_consume_pages, sizeof(*consume_ppns),
453 			  GFP_KERNEL);
454 	if (!consume_ppns) {
455 		kfree(produce_ppns);
456 		return VMCI_ERROR_NO_MEM;
457 	}
458 
459 	for (i = 0; i < num_produce_pages; i++)
460 		produce_ppns[i] =
461 			produce_q->kernel_if->u.g.pas[i] >> PAGE_SHIFT;
462 
463 	for (i = 0; i < num_consume_pages; i++)
464 		consume_ppns[i] =
465 			consume_q->kernel_if->u.g.pas[i] >> PAGE_SHIFT;
466 
467 	ppn_set->num_produce_pages = num_produce_pages;
468 	ppn_set->num_consume_pages = num_consume_pages;
469 	ppn_set->produce_ppns = produce_ppns;
470 	ppn_set->consume_ppns = consume_ppns;
471 	ppn_set->initialized = true;
472 	return VMCI_SUCCESS;
473 }
474 
475 /*
476  * Frees the two list of PPNs for a queue pair.
477  */
478 static void qp_free_ppn_set(struct ppn_set *ppn_set)
479 {
480 	if (ppn_set->initialized) {
481 		/* Do not call these functions on NULL inputs. */
482 		kfree(ppn_set->produce_ppns);
483 		kfree(ppn_set->consume_ppns);
484 	}
485 	memset(ppn_set, 0, sizeof(*ppn_set));
486 }
487 
488 /*
489  * Populates the list of PPNs in the hypercall structure with the PPNS
490  * of the produce queue and the consume queue.
491  */
492 static int qp_populate_ppn_set(u8 *call_buf, const struct ppn_set *ppn_set)
493 {
494 	if (vmci_use_ppn64()) {
495 		memcpy(call_buf, ppn_set->produce_ppns,
496 		       ppn_set->num_produce_pages *
497 		       sizeof(*ppn_set->produce_ppns));
498 		memcpy(call_buf +
499 		       ppn_set->num_produce_pages *
500 		       sizeof(*ppn_set->produce_ppns),
501 		       ppn_set->consume_ppns,
502 		       ppn_set->num_consume_pages *
503 		       sizeof(*ppn_set->consume_ppns));
504 	} else {
505 		int i;
506 		u32 *ppns = (u32 *) call_buf;
507 
508 		for (i = 0; i < ppn_set->num_produce_pages; i++)
509 			ppns[i] = (u32) ppn_set->produce_ppns[i];
510 
511 		ppns = &ppns[ppn_set->num_produce_pages];
512 
513 		for (i = 0; i < ppn_set->num_consume_pages; i++)
514 			ppns[i] = (u32) ppn_set->consume_ppns[i];
515 	}
516 
517 	return VMCI_SUCCESS;
518 }
519 
520 /*
521  * Allocates kernel VA space of specified size plus space for the queue
522  * and kernel interface.  This is different from the guest queue allocator,
523  * because we do not allocate our own queue header/data pages here but
524  * share those of the guest.
525  */
526 static struct vmci_queue *qp_host_alloc_queue(u64 size)
527 {
528 	struct vmci_queue *queue;
529 	size_t queue_page_size;
530 	u64 num_pages;
531 	const size_t queue_size = sizeof(*queue) + sizeof(*(queue->kernel_if));
532 
533 	if (size > min_t(size_t, VMCI_MAX_GUEST_QP_MEMORY, SIZE_MAX - PAGE_SIZE))
534 		return NULL;
535 	num_pages = DIV_ROUND_UP(size, PAGE_SIZE) + 1;
536 	if (num_pages > (SIZE_MAX - queue_size) /
537 		 sizeof(*queue->kernel_if->u.h.page))
538 		return NULL;
539 
540 	queue_page_size = num_pages * sizeof(*queue->kernel_if->u.h.page);
541 
542 	if (queue_size + queue_page_size > KMALLOC_MAX_SIZE)
543 		return NULL;
544 
545 	queue = kzalloc(queue_size + queue_page_size, GFP_KERNEL);
546 	if (queue) {
547 		queue->q_header = NULL;
548 		queue->saved_header = NULL;
549 		queue->kernel_if = (struct vmci_queue_kern_if *)(queue + 1);
550 		queue->kernel_if->host = true;
551 		queue->kernel_if->mutex = NULL;
552 		queue->kernel_if->num_pages = num_pages;
553 		queue->kernel_if->u.h.header_page =
554 		    (struct page **)((u8 *)queue + queue_size);
555 		queue->kernel_if->u.h.page =
556 			&queue->kernel_if->u.h.header_page[1];
557 	}
558 
559 	return queue;
560 }
561 
562 /*
563  * Frees kernel memory for a given queue (header plus translation
564  * structure).
565  */
566 static void qp_host_free_queue(struct vmci_queue *queue, u64 queue_size)
567 {
568 	kfree(queue);
569 }
570 
571 /*
572  * Initialize the mutex for the pair of queues.  This mutex is used to
573  * protect the q_header and the buffer from changing out from under any
574  * users of either queue.  Of course, it's only any good if the mutexes
575  * are actually acquired.  Queue structure must lie on non-paged memory
576  * or we cannot guarantee access to the mutex.
577  */
578 static void qp_init_queue_mutex(struct vmci_queue *produce_q,
579 				struct vmci_queue *consume_q)
580 {
581 	/*
582 	 * Only the host queue has shared state - the guest queues do not
583 	 * need to synchronize access using a queue mutex.
584 	 */
585 
586 	if (produce_q->kernel_if->host) {
587 		produce_q->kernel_if->mutex = &produce_q->kernel_if->__mutex;
588 		consume_q->kernel_if->mutex = &produce_q->kernel_if->__mutex;
589 		mutex_init(produce_q->kernel_if->mutex);
590 	}
591 }
592 
593 /*
594  * Cleans up the mutex for the pair of queues.
595  */
596 static void qp_cleanup_queue_mutex(struct vmci_queue *produce_q,
597 				   struct vmci_queue *consume_q)
598 {
599 	if (produce_q->kernel_if->host) {
600 		produce_q->kernel_if->mutex = NULL;
601 		consume_q->kernel_if->mutex = NULL;
602 	}
603 }
604 
605 /*
606  * Acquire the mutex for the queue.  Note that the produce_q and
607  * the consume_q share a mutex.  So, only one of the two need to
608  * be passed in to this routine.  Either will work just fine.
609  */
610 static void qp_acquire_queue_mutex(struct vmci_queue *queue)
611 {
612 	if (queue->kernel_if->host)
613 		mutex_lock(queue->kernel_if->mutex);
614 }
615 
616 /*
617  * Release the mutex for the queue.  Note that the produce_q and
618  * the consume_q share a mutex.  So, only one of the two need to
619  * be passed in to this routine.  Either will work just fine.
620  */
621 static void qp_release_queue_mutex(struct vmci_queue *queue)
622 {
623 	if (queue->kernel_if->host)
624 		mutex_unlock(queue->kernel_if->mutex);
625 }
626 
627 /*
628  * Helper function to release pages in the PageStoreAttachInfo
629  * previously obtained using get_user_pages.
630  */
631 static void qp_release_pages(struct page **pages,
632 			     u64 num_pages, bool dirty)
633 {
634 	int i;
635 
636 	for (i = 0; i < num_pages; i++) {
637 		if (dirty)
638 			set_page_dirty_lock(pages[i]);
639 
640 		put_page(pages[i]);
641 		pages[i] = NULL;
642 	}
643 }
644 
645 /*
646  * Lock the user pages referenced by the {produce,consume}Buffer
647  * struct into memory and populate the {produce,consume}Pages
648  * arrays in the attach structure with them.
649  */
650 static int qp_host_get_user_memory(u64 produce_uva,
651 				   u64 consume_uva,
652 				   struct vmci_queue *produce_q,
653 				   struct vmci_queue *consume_q)
654 {
655 	int retval;
656 	int err = VMCI_SUCCESS;
657 
658 	retval = get_user_pages_fast((uintptr_t) produce_uva,
659 				     produce_q->kernel_if->num_pages,
660 				     FOLL_WRITE,
661 				     produce_q->kernel_if->u.h.header_page);
662 	if (retval < (int)produce_q->kernel_if->num_pages) {
663 		pr_debug("get_user_pages_fast(produce) failed (retval=%d)",
664 			retval);
665 		if (retval > 0)
666 			qp_release_pages(produce_q->kernel_if->u.h.header_page,
667 					retval, false);
668 		err = VMCI_ERROR_NO_MEM;
669 		goto out;
670 	}
671 
672 	retval = get_user_pages_fast((uintptr_t) consume_uva,
673 				     consume_q->kernel_if->num_pages,
674 				     FOLL_WRITE,
675 				     consume_q->kernel_if->u.h.header_page);
676 	if (retval < (int)consume_q->kernel_if->num_pages) {
677 		pr_debug("get_user_pages_fast(consume) failed (retval=%d)",
678 			retval);
679 		if (retval > 0)
680 			qp_release_pages(consume_q->kernel_if->u.h.header_page,
681 					retval, false);
682 		qp_release_pages(produce_q->kernel_if->u.h.header_page,
683 				 produce_q->kernel_if->num_pages, false);
684 		err = VMCI_ERROR_NO_MEM;
685 	}
686 
687  out:
688 	return err;
689 }
690 
691 /*
692  * Registers the specification of the user pages used for backing a queue
693  * pair. Enough information to map in pages is stored in the OS specific
694  * part of the struct vmci_queue structure.
695  */
696 static int qp_host_register_user_memory(struct vmci_qp_page_store *page_store,
697 					struct vmci_queue *produce_q,
698 					struct vmci_queue *consume_q)
699 {
700 	u64 produce_uva;
701 	u64 consume_uva;
702 
703 	/*
704 	 * The new style and the old style mapping only differs in
705 	 * that we either get a single or two UVAs, so we split the
706 	 * single UVA range at the appropriate spot.
707 	 */
708 	produce_uva = page_store->pages;
709 	consume_uva = page_store->pages +
710 	    produce_q->kernel_if->num_pages * PAGE_SIZE;
711 	return qp_host_get_user_memory(produce_uva, consume_uva, produce_q,
712 				       consume_q);
713 }
714 
715 /*
716  * Releases and removes the references to user pages stored in the attach
717  * struct.  Pages are released from the page cache and may become
718  * swappable again.
719  */
720 static void qp_host_unregister_user_memory(struct vmci_queue *produce_q,
721 					   struct vmci_queue *consume_q)
722 {
723 	qp_release_pages(produce_q->kernel_if->u.h.header_page,
724 			 produce_q->kernel_if->num_pages, true);
725 	memset(produce_q->kernel_if->u.h.header_page, 0,
726 	       sizeof(*produce_q->kernel_if->u.h.header_page) *
727 	       produce_q->kernel_if->num_pages);
728 	qp_release_pages(consume_q->kernel_if->u.h.header_page,
729 			 consume_q->kernel_if->num_pages, true);
730 	memset(consume_q->kernel_if->u.h.header_page, 0,
731 	       sizeof(*consume_q->kernel_if->u.h.header_page) *
732 	       consume_q->kernel_if->num_pages);
733 }
734 
735 /*
736  * Once qp_host_register_user_memory has been performed on a
737  * queue, the queue pair headers can be mapped into the
738  * kernel. Once mapped, they must be unmapped with
739  * qp_host_unmap_queues prior to calling
740  * qp_host_unregister_user_memory.
741  * Pages are pinned.
742  */
743 static int qp_host_map_queues(struct vmci_queue *produce_q,
744 			      struct vmci_queue *consume_q)
745 {
746 	int result;
747 
748 	if (!produce_q->q_header || !consume_q->q_header) {
749 		struct page *headers[2];
750 
751 		if (produce_q->q_header != consume_q->q_header)
752 			return VMCI_ERROR_QUEUEPAIR_MISMATCH;
753 
754 		if (produce_q->kernel_if->u.h.header_page == NULL ||
755 		    *produce_q->kernel_if->u.h.header_page == NULL)
756 			return VMCI_ERROR_UNAVAILABLE;
757 
758 		headers[0] = *produce_q->kernel_if->u.h.header_page;
759 		headers[1] = *consume_q->kernel_if->u.h.header_page;
760 
761 		produce_q->q_header = vmap(headers, 2, VM_MAP, PAGE_KERNEL);
762 		if (produce_q->q_header != NULL) {
763 			consume_q->q_header =
764 			    (struct vmci_queue_header *)((u8 *)
765 							 produce_q->q_header +
766 							 PAGE_SIZE);
767 			result = VMCI_SUCCESS;
768 		} else {
769 			pr_warn("vmap failed\n");
770 			result = VMCI_ERROR_NO_MEM;
771 		}
772 	} else {
773 		result = VMCI_SUCCESS;
774 	}
775 
776 	return result;
777 }
778 
779 /*
780  * Unmaps previously mapped queue pair headers from the kernel.
781  * Pages are unpinned.
782  */
783 static int qp_host_unmap_queues(u32 gid,
784 				struct vmci_queue *produce_q,
785 				struct vmci_queue *consume_q)
786 {
787 	if (produce_q->q_header) {
788 		if (produce_q->q_header < consume_q->q_header)
789 			vunmap(produce_q->q_header);
790 		else
791 			vunmap(consume_q->q_header);
792 
793 		produce_q->q_header = NULL;
794 		consume_q->q_header = NULL;
795 	}
796 
797 	return VMCI_SUCCESS;
798 }
799 
800 /*
801  * Finds the entry in the list corresponding to a given handle. Assumes
802  * that the list is locked.
803  */
804 static struct qp_entry *qp_list_find(struct qp_list *qp_list,
805 				     struct vmci_handle handle)
806 {
807 	struct qp_entry *entry;
808 
809 	if (vmci_handle_is_invalid(handle))
810 		return NULL;
811 
812 	list_for_each_entry(entry, &qp_list->head, list_item) {
813 		if (vmci_handle_is_equal(entry->handle, handle))
814 			return entry;
815 	}
816 
817 	return NULL;
818 }
819 
820 /*
821  * Finds the entry in the list corresponding to a given handle.
822  */
823 static struct qp_guest_endpoint *
824 qp_guest_handle_to_entry(struct vmci_handle handle)
825 {
826 	struct qp_guest_endpoint *entry;
827 	struct qp_entry *qp = qp_list_find(&qp_guest_endpoints, handle);
828 
829 	entry = qp ? container_of(
830 		qp, struct qp_guest_endpoint, qp) : NULL;
831 	return entry;
832 }
833 
834 /*
835  * Finds the entry in the list corresponding to a given handle.
836  */
837 static struct qp_broker_entry *
838 qp_broker_handle_to_entry(struct vmci_handle handle)
839 {
840 	struct qp_broker_entry *entry;
841 	struct qp_entry *qp = qp_list_find(&qp_broker_list, handle);
842 
843 	entry = qp ? container_of(
844 		qp, struct qp_broker_entry, qp) : NULL;
845 	return entry;
846 }
847 
848 /*
849  * Dispatches a queue pair event message directly into the local event
850  * queue.
851  */
852 static int qp_notify_peer_local(bool attach, struct vmci_handle handle)
853 {
854 	u32 context_id = vmci_get_context_id();
855 	struct vmci_event_qp ev;
856 
857 	memset(&ev, 0, sizeof(ev));
858 	ev.msg.hdr.dst = vmci_make_handle(context_id, VMCI_EVENT_HANDLER);
859 	ev.msg.hdr.src = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
860 					  VMCI_CONTEXT_RESOURCE_ID);
861 	ev.msg.hdr.payload_size = sizeof(ev) - sizeof(ev.msg.hdr);
862 	ev.msg.event_data.event =
863 	    attach ? VMCI_EVENT_QP_PEER_ATTACH : VMCI_EVENT_QP_PEER_DETACH;
864 	ev.payload.peer_id = context_id;
865 	ev.payload.handle = handle;
866 
867 	return vmci_event_dispatch(&ev.msg.hdr);
868 }
869 
870 /*
871  * Allocates and initializes a qp_guest_endpoint structure.
872  * Allocates a queue_pair rid (and handle) iff the given entry has
873  * an invalid handle.  0 through VMCI_RESERVED_RESOURCE_ID_MAX
874  * are reserved handles.  Assumes that the QP list mutex is held
875  * by the caller.
876  */
877 static struct qp_guest_endpoint *
878 qp_guest_endpoint_create(struct vmci_handle handle,
879 			 u32 peer,
880 			 u32 flags,
881 			 u64 produce_size,
882 			 u64 consume_size,
883 			 void *produce_q,
884 			 void *consume_q)
885 {
886 	int result;
887 	struct qp_guest_endpoint *entry;
888 	/* One page each for the queue headers. */
889 	const u64 num_ppns = DIV_ROUND_UP(produce_size, PAGE_SIZE) +
890 	    DIV_ROUND_UP(consume_size, PAGE_SIZE) + 2;
891 
892 	if (vmci_handle_is_invalid(handle)) {
893 		u32 context_id = vmci_get_context_id();
894 
895 		handle = vmci_make_handle(context_id, VMCI_INVALID_ID);
896 	}
897 
898 	entry = kzalloc(sizeof(*entry), GFP_KERNEL);
899 	if (entry) {
900 		entry->qp.peer = peer;
901 		entry->qp.flags = flags;
902 		entry->qp.produce_size = produce_size;
903 		entry->qp.consume_size = consume_size;
904 		entry->qp.ref_count = 0;
905 		entry->num_ppns = num_ppns;
906 		entry->produce_q = produce_q;
907 		entry->consume_q = consume_q;
908 		INIT_LIST_HEAD(&entry->qp.list_item);
909 
910 		/* Add resource obj */
911 		result = vmci_resource_add(&entry->resource,
912 					   VMCI_RESOURCE_TYPE_QPAIR_GUEST,
913 					   handle);
914 		entry->qp.handle = vmci_resource_handle(&entry->resource);
915 		if ((result != VMCI_SUCCESS) ||
916 		    qp_list_find(&qp_guest_endpoints, entry->qp.handle)) {
917 			pr_warn("Failed to add new resource (handle=0x%x:0x%x), error: %d",
918 				handle.context, handle.resource, result);
919 			kfree(entry);
920 			entry = NULL;
921 		}
922 	}
923 	return entry;
924 }
925 
926 /*
927  * Frees a qp_guest_endpoint structure.
928  */
929 static void qp_guest_endpoint_destroy(struct qp_guest_endpoint *entry)
930 {
931 	qp_free_ppn_set(&entry->ppn_set);
932 	qp_cleanup_queue_mutex(entry->produce_q, entry->consume_q);
933 	qp_free_queue(entry->produce_q, entry->qp.produce_size);
934 	qp_free_queue(entry->consume_q, entry->qp.consume_size);
935 	/* Unlink from resource hash table and free callback */
936 	vmci_resource_remove(&entry->resource);
937 
938 	kfree(entry);
939 }
940 
941 /*
942  * Helper to make a queue_pairAlloc hypercall when the driver is
943  * supporting a guest device.
944  */
945 static int qp_alloc_hypercall(const struct qp_guest_endpoint *entry)
946 {
947 	struct vmci_qp_alloc_msg *alloc_msg;
948 	size_t msg_size;
949 	size_t ppn_size;
950 	int result;
951 
952 	if (!entry || entry->num_ppns <= 2)
953 		return VMCI_ERROR_INVALID_ARGS;
954 
955 	ppn_size = vmci_use_ppn64() ? sizeof(u64) : sizeof(u32);
956 	msg_size = sizeof(*alloc_msg) +
957 	    (size_t) entry->num_ppns * ppn_size;
958 	alloc_msg = kmalloc(msg_size, GFP_KERNEL);
959 	if (!alloc_msg)
960 		return VMCI_ERROR_NO_MEM;
961 
962 	alloc_msg->hdr.dst = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
963 					      VMCI_QUEUEPAIR_ALLOC);
964 	alloc_msg->hdr.src = VMCI_ANON_SRC_HANDLE;
965 	alloc_msg->hdr.payload_size = msg_size - VMCI_DG_HEADERSIZE;
966 	alloc_msg->handle = entry->qp.handle;
967 	alloc_msg->peer = entry->qp.peer;
968 	alloc_msg->flags = entry->qp.flags;
969 	alloc_msg->produce_size = entry->qp.produce_size;
970 	alloc_msg->consume_size = entry->qp.consume_size;
971 	alloc_msg->num_ppns = entry->num_ppns;
972 
973 	result = qp_populate_ppn_set((u8 *)alloc_msg + sizeof(*alloc_msg),
974 				     &entry->ppn_set);
975 	if (result == VMCI_SUCCESS)
976 		result = vmci_send_datagram(&alloc_msg->hdr);
977 
978 	kfree(alloc_msg);
979 
980 	return result;
981 }
982 
983 /*
984  * Helper to make a queue_pairDetach hypercall when the driver is
985  * supporting a guest device.
986  */
987 static int qp_detatch_hypercall(struct vmci_handle handle)
988 {
989 	struct vmci_qp_detach_msg detach_msg;
990 
991 	detach_msg.hdr.dst = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
992 					      VMCI_QUEUEPAIR_DETACH);
993 	detach_msg.hdr.src = VMCI_ANON_SRC_HANDLE;
994 	detach_msg.hdr.payload_size = sizeof(handle);
995 	detach_msg.handle = handle;
996 
997 	return vmci_send_datagram(&detach_msg.hdr);
998 }
999 
1000 /*
1001  * Adds the given entry to the list. Assumes that the list is locked.
1002  */
1003 static void qp_list_add_entry(struct qp_list *qp_list, struct qp_entry *entry)
1004 {
1005 	if (entry)
1006 		list_add(&entry->list_item, &qp_list->head);
1007 }
1008 
1009 /*
1010  * Removes the given entry from the list. Assumes that the list is locked.
1011  */
1012 static void qp_list_remove_entry(struct qp_list *qp_list,
1013 				 struct qp_entry *entry)
1014 {
1015 	if (entry)
1016 		list_del(&entry->list_item);
1017 }
1018 
1019 /*
1020  * Helper for VMCI queue_pair detach interface. Frees the physical
1021  * pages for the queue pair.
1022  */
1023 static int qp_detatch_guest_work(struct vmci_handle handle)
1024 {
1025 	int result;
1026 	struct qp_guest_endpoint *entry;
1027 	u32 ref_count = ~0;	/* To avoid compiler warning below */
1028 
1029 	mutex_lock(&qp_guest_endpoints.mutex);
1030 
1031 	entry = qp_guest_handle_to_entry(handle);
1032 	if (!entry) {
1033 		mutex_unlock(&qp_guest_endpoints.mutex);
1034 		return VMCI_ERROR_NOT_FOUND;
1035 	}
1036 
1037 	if (entry->qp.flags & VMCI_QPFLAG_LOCAL) {
1038 		result = VMCI_SUCCESS;
1039 
1040 		if (entry->qp.ref_count > 1) {
1041 			result = qp_notify_peer_local(false, handle);
1042 			/*
1043 			 * We can fail to notify a local queuepair
1044 			 * because we can't allocate.  We still want
1045 			 * to release the entry if that happens, so
1046 			 * don't bail out yet.
1047 			 */
1048 		}
1049 	} else {
1050 		result = qp_detatch_hypercall(handle);
1051 		if (result < VMCI_SUCCESS) {
1052 			/*
1053 			 * We failed to notify a non-local queuepair.
1054 			 * That other queuepair might still be
1055 			 * accessing the shared memory, so don't
1056 			 * release the entry yet.  It will get cleaned
1057 			 * up by VMCIqueue_pair_Exit() if necessary
1058 			 * (assuming we are going away, otherwise why
1059 			 * did this fail?).
1060 			 */
1061 
1062 			mutex_unlock(&qp_guest_endpoints.mutex);
1063 			return result;
1064 		}
1065 	}
1066 
1067 	/*
1068 	 * If we get here then we either failed to notify a local queuepair, or
1069 	 * we succeeded in all cases.  Release the entry if required.
1070 	 */
1071 
1072 	entry->qp.ref_count--;
1073 	if (entry->qp.ref_count == 0)
1074 		qp_list_remove_entry(&qp_guest_endpoints, &entry->qp);
1075 
1076 	/* If we didn't remove the entry, this could change once we unlock. */
1077 	if (entry)
1078 		ref_count = entry->qp.ref_count;
1079 
1080 	mutex_unlock(&qp_guest_endpoints.mutex);
1081 
1082 	if (ref_count == 0)
1083 		qp_guest_endpoint_destroy(entry);
1084 
1085 	return result;
1086 }
1087 
1088 /*
1089  * This functions handles the actual allocation of a VMCI queue
1090  * pair guest endpoint. Allocates physical pages for the queue
1091  * pair. It makes OS dependent calls through generic wrappers.
1092  */
1093 static int qp_alloc_guest_work(struct vmci_handle *handle,
1094 			       struct vmci_queue **produce_q,
1095 			       u64 produce_size,
1096 			       struct vmci_queue **consume_q,
1097 			       u64 consume_size,
1098 			       u32 peer,
1099 			       u32 flags,
1100 			       u32 priv_flags)
1101 {
1102 	const u64 num_produce_pages =
1103 	    DIV_ROUND_UP(produce_size, PAGE_SIZE) + 1;
1104 	const u64 num_consume_pages =
1105 	    DIV_ROUND_UP(consume_size, PAGE_SIZE) + 1;
1106 	void *my_produce_q = NULL;
1107 	void *my_consume_q = NULL;
1108 	int result;
1109 	struct qp_guest_endpoint *queue_pair_entry = NULL;
1110 
1111 	if (priv_flags != VMCI_NO_PRIVILEGE_FLAGS)
1112 		return VMCI_ERROR_NO_ACCESS;
1113 
1114 	mutex_lock(&qp_guest_endpoints.mutex);
1115 
1116 	queue_pair_entry = qp_guest_handle_to_entry(*handle);
1117 	if (queue_pair_entry) {
1118 		if (queue_pair_entry->qp.flags & VMCI_QPFLAG_LOCAL) {
1119 			/* Local attach case. */
1120 			if (queue_pair_entry->qp.ref_count > 1) {
1121 				pr_devel("Error attempting to attach more than once\n");
1122 				result = VMCI_ERROR_UNAVAILABLE;
1123 				goto error_keep_entry;
1124 			}
1125 
1126 			if (queue_pair_entry->qp.produce_size != consume_size ||
1127 			    queue_pair_entry->qp.consume_size !=
1128 			    produce_size ||
1129 			    queue_pair_entry->qp.flags !=
1130 			    (flags & ~VMCI_QPFLAG_ATTACH_ONLY)) {
1131 				pr_devel("Error mismatched queue pair in local attach\n");
1132 				result = VMCI_ERROR_QUEUEPAIR_MISMATCH;
1133 				goto error_keep_entry;
1134 			}
1135 
1136 			/*
1137 			 * Do a local attach.  We swap the consume and
1138 			 * produce queues for the attacher and deliver
1139 			 * an attach event.
1140 			 */
1141 			result = qp_notify_peer_local(true, *handle);
1142 			if (result < VMCI_SUCCESS)
1143 				goto error_keep_entry;
1144 
1145 			my_produce_q = queue_pair_entry->consume_q;
1146 			my_consume_q = queue_pair_entry->produce_q;
1147 			goto out;
1148 		}
1149 
1150 		result = VMCI_ERROR_ALREADY_EXISTS;
1151 		goto error_keep_entry;
1152 	}
1153 
1154 	my_produce_q = qp_alloc_queue(produce_size, flags);
1155 	if (!my_produce_q) {
1156 		pr_warn("Error allocating pages for produce queue\n");
1157 		result = VMCI_ERROR_NO_MEM;
1158 		goto error;
1159 	}
1160 
1161 	my_consume_q = qp_alloc_queue(consume_size, flags);
1162 	if (!my_consume_q) {
1163 		pr_warn("Error allocating pages for consume queue\n");
1164 		result = VMCI_ERROR_NO_MEM;
1165 		goto error;
1166 	}
1167 
1168 	queue_pair_entry = qp_guest_endpoint_create(*handle, peer, flags,
1169 						    produce_size, consume_size,
1170 						    my_produce_q, my_consume_q);
1171 	if (!queue_pair_entry) {
1172 		pr_warn("Error allocating memory in %s\n", __func__);
1173 		result = VMCI_ERROR_NO_MEM;
1174 		goto error;
1175 	}
1176 
1177 	result = qp_alloc_ppn_set(my_produce_q, num_produce_pages, my_consume_q,
1178 				  num_consume_pages,
1179 				  &queue_pair_entry->ppn_set);
1180 	if (result < VMCI_SUCCESS) {
1181 		pr_warn("qp_alloc_ppn_set failed\n");
1182 		goto error;
1183 	}
1184 
1185 	/*
1186 	 * It's only necessary to notify the host if this queue pair will be
1187 	 * attached to from another context.
1188 	 */
1189 	if (queue_pair_entry->qp.flags & VMCI_QPFLAG_LOCAL) {
1190 		/* Local create case. */
1191 		u32 context_id = vmci_get_context_id();
1192 
1193 		/*
1194 		 * Enforce similar checks on local queue pairs as we
1195 		 * do for regular ones.  The handle's context must
1196 		 * match the creator or attacher context id (here they
1197 		 * are both the current context id) and the
1198 		 * attach-only flag cannot exist during create.  We
1199 		 * also ensure specified peer is this context or an
1200 		 * invalid one.
1201 		 */
1202 		if (queue_pair_entry->qp.handle.context != context_id ||
1203 		    (queue_pair_entry->qp.peer != VMCI_INVALID_ID &&
1204 		     queue_pair_entry->qp.peer != context_id)) {
1205 			result = VMCI_ERROR_NO_ACCESS;
1206 			goto error;
1207 		}
1208 
1209 		if (queue_pair_entry->qp.flags & VMCI_QPFLAG_ATTACH_ONLY) {
1210 			result = VMCI_ERROR_NOT_FOUND;
1211 			goto error;
1212 		}
1213 	} else {
1214 		result = qp_alloc_hypercall(queue_pair_entry);
1215 		if (result < VMCI_SUCCESS) {
1216 			pr_devel("qp_alloc_hypercall result = %d\n", result);
1217 			goto error;
1218 		}
1219 	}
1220 
1221 	qp_init_queue_mutex((struct vmci_queue *)my_produce_q,
1222 			    (struct vmci_queue *)my_consume_q);
1223 
1224 	qp_list_add_entry(&qp_guest_endpoints, &queue_pair_entry->qp);
1225 
1226  out:
1227 	queue_pair_entry->qp.ref_count++;
1228 	*handle = queue_pair_entry->qp.handle;
1229 	*produce_q = (struct vmci_queue *)my_produce_q;
1230 	*consume_q = (struct vmci_queue *)my_consume_q;
1231 
1232 	/*
1233 	 * We should initialize the queue pair header pages on a local
1234 	 * queue pair create.  For non-local queue pairs, the
1235 	 * hypervisor initializes the header pages in the create step.
1236 	 */
1237 	if ((queue_pair_entry->qp.flags & VMCI_QPFLAG_LOCAL) &&
1238 	    queue_pair_entry->qp.ref_count == 1) {
1239 		vmci_q_header_init((*produce_q)->q_header, *handle);
1240 		vmci_q_header_init((*consume_q)->q_header, *handle);
1241 	}
1242 
1243 	mutex_unlock(&qp_guest_endpoints.mutex);
1244 
1245 	return VMCI_SUCCESS;
1246 
1247  error:
1248 	mutex_unlock(&qp_guest_endpoints.mutex);
1249 	if (queue_pair_entry) {
1250 		/* The queues will be freed inside the destroy routine. */
1251 		qp_guest_endpoint_destroy(queue_pair_entry);
1252 	} else {
1253 		qp_free_queue(my_produce_q, produce_size);
1254 		qp_free_queue(my_consume_q, consume_size);
1255 	}
1256 	return result;
1257 
1258  error_keep_entry:
1259 	/* This path should only be used when an existing entry was found. */
1260 	mutex_unlock(&qp_guest_endpoints.mutex);
1261 	return result;
1262 }
1263 
1264 /*
1265  * The first endpoint issuing a queue pair allocation will create the state
1266  * of the queue pair in the queue pair broker.
1267  *
1268  * If the creator is a guest, it will associate a VMX virtual address range
1269  * with the queue pair as specified by the page_store. For compatibility with
1270  * older VMX'en, that would use a separate step to set the VMX virtual
1271  * address range, the virtual address range can be registered later using
1272  * vmci_qp_broker_set_page_store. In that case, a page_store of NULL should be
1273  * used.
1274  *
1275  * If the creator is the host, a page_store of NULL should be used as well,
1276  * since the host is not able to supply a page store for the queue pair.
1277  *
1278  * For older VMX and host callers, the queue pair will be created in the
1279  * VMCIQPB_CREATED_NO_MEM state, and for current VMX callers, it will be
1280  * created in VMCOQPB_CREATED_MEM state.
1281  */
1282 static int qp_broker_create(struct vmci_handle handle,
1283 			    u32 peer,
1284 			    u32 flags,
1285 			    u32 priv_flags,
1286 			    u64 produce_size,
1287 			    u64 consume_size,
1288 			    struct vmci_qp_page_store *page_store,
1289 			    struct vmci_ctx *context,
1290 			    vmci_event_release_cb wakeup_cb,
1291 			    void *client_data, struct qp_broker_entry **ent)
1292 {
1293 	struct qp_broker_entry *entry = NULL;
1294 	const u32 context_id = vmci_ctx_get_id(context);
1295 	bool is_local = flags & VMCI_QPFLAG_LOCAL;
1296 	int result;
1297 	u64 guest_produce_size;
1298 	u64 guest_consume_size;
1299 
1300 	/* Do not create if the caller asked not to. */
1301 	if (flags & VMCI_QPFLAG_ATTACH_ONLY)
1302 		return VMCI_ERROR_NOT_FOUND;
1303 
1304 	/*
1305 	 * Creator's context ID should match handle's context ID or the creator
1306 	 * must allow the context in handle's context ID as the "peer".
1307 	 */
1308 	if (handle.context != context_id && handle.context != peer)
1309 		return VMCI_ERROR_NO_ACCESS;
1310 
1311 	if (VMCI_CONTEXT_IS_VM(context_id) && VMCI_CONTEXT_IS_VM(peer))
1312 		return VMCI_ERROR_DST_UNREACHABLE;
1313 
1314 	/*
1315 	 * Creator's context ID for local queue pairs should match the
1316 	 * peer, if a peer is specified.
1317 	 */
1318 	if (is_local && peer != VMCI_INVALID_ID && context_id != peer)
1319 		return VMCI_ERROR_NO_ACCESS;
1320 
1321 	entry = kzalloc(sizeof(*entry), GFP_ATOMIC);
1322 	if (!entry)
1323 		return VMCI_ERROR_NO_MEM;
1324 
1325 	if (vmci_ctx_get_id(context) == VMCI_HOST_CONTEXT_ID && !is_local) {
1326 		/*
1327 		 * The queue pair broker entry stores values from the guest
1328 		 * point of view, so a creating host side endpoint should swap
1329 		 * produce and consume values -- unless it is a local queue
1330 		 * pair, in which case no swapping is necessary, since the local
1331 		 * attacher will swap queues.
1332 		 */
1333 
1334 		guest_produce_size = consume_size;
1335 		guest_consume_size = produce_size;
1336 	} else {
1337 		guest_produce_size = produce_size;
1338 		guest_consume_size = consume_size;
1339 	}
1340 
1341 	entry->qp.handle = handle;
1342 	entry->qp.peer = peer;
1343 	entry->qp.flags = flags;
1344 	entry->qp.produce_size = guest_produce_size;
1345 	entry->qp.consume_size = guest_consume_size;
1346 	entry->qp.ref_count = 1;
1347 	entry->create_id = context_id;
1348 	entry->attach_id = VMCI_INVALID_ID;
1349 	entry->state = VMCIQPB_NEW;
1350 	entry->require_trusted_attach =
1351 	    !!(context->priv_flags & VMCI_PRIVILEGE_FLAG_RESTRICTED);
1352 	entry->created_by_trusted =
1353 	    !!(priv_flags & VMCI_PRIVILEGE_FLAG_TRUSTED);
1354 	entry->vmci_page_files = false;
1355 	entry->wakeup_cb = wakeup_cb;
1356 	entry->client_data = client_data;
1357 	entry->produce_q = qp_host_alloc_queue(guest_produce_size);
1358 	if (entry->produce_q == NULL) {
1359 		result = VMCI_ERROR_NO_MEM;
1360 		goto error;
1361 	}
1362 	entry->consume_q = qp_host_alloc_queue(guest_consume_size);
1363 	if (entry->consume_q == NULL) {
1364 		result = VMCI_ERROR_NO_MEM;
1365 		goto error;
1366 	}
1367 
1368 	qp_init_queue_mutex(entry->produce_q, entry->consume_q);
1369 
1370 	INIT_LIST_HEAD(&entry->qp.list_item);
1371 
1372 	if (is_local) {
1373 		u8 *tmp;
1374 
1375 		entry->local_mem = kcalloc(QPE_NUM_PAGES(entry->qp),
1376 					   PAGE_SIZE, GFP_KERNEL);
1377 		if (entry->local_mem == NULL) {
1378 			result = VMCI_ERROR_NO_MEM;
1379 			goto error;
1380 		}
1381 		entry->state = VMCIQPB_CREATED_MEM;
1382 		entry->produce_q->q_header = entry->local_mem;
1383 		tmp = (u8 *)entry->local_mem + PAGE_SIZE *
1384 		    (DIV_ROUND_UP(entry->qp.produce_size, PAGE_SIZE) + 1);
1385 		entry->consume_q->q_header = (struct vmci_queue_header *)tmp;
1386 	} else if (page_store) {
1387 		/*
1388 		 * The VMX already initialized the queue pair headers, so no
1389 		 * need for the kernel side to do that.
1390 		 */
1391 		result = qp_host_register_user_memory(page_store,
1392 						      entry->produce_q,
1393 						      entry->consume_q);
1394 		if (result < VMCI_SUCCESS)
1395 			goto error;
1396 
1397 		entry->state = VMCIQPB_CREATED_MEM;
1398 	} else {
1399 		/*
1400 		 * A create without a page_store may be either a host
1401 		 * side create (in which case we are waiting for the
1402 		 * guest side to supply the memory) or an old style
1403 		 * queue pair create (in which case we will expect a
1404 		 * set page store call as the next step).
1405 		 */
1406 		entry->state = VMCIQPB_CREATED_NO_MEM;
1407 	}
1408 
1409 	qp_list_add_entry(&qp_broker_list, &entry->qp);
1410 	if (ent != NULL)
1411 		*ent = entry;
1412 
1413 	/* Add to resource obj */
1414 	result = vmci_resource_add(&entry->resource,
1415 				   VMCI_RESOURCE_TYPE_QPAIR_HOST,
1416 				   handle);
1417 	if (result != VMCI_SUCCESS) {
1418 		pr_warn("Failed to add new resource (handle=0x%x:0x%x), error: %d",
1419 			handle.context, handle.resource, result);
1420 		goto error;
1421 	}
1422 
1423 	entry->qp.handle = vmci_resource_handle(&entry->resource);
1424 	if (is_local) {
1425 		vmci_q_header_init(entry->produce_q->q_header,
1426 				   entry->qp.handle);
1427 		vmci_q_header_init(entry->consume_q->q_header,
1428 				   entry->qp.handle);
1429 	}
1430 
1431 	vmci_ctx_qp_create(context, entry->qp.handle);
1432 
1433 	return VMCI_SUCCESS;
1434 
1435  error:
1436 	if (entry != NULL) {
1437 		qp_host_free_queue(entry->produce_q, guest_produce_size);
1438 		qp_host_free_queue(entry->consume_q, guest_consume_size);
1439 		kfree(entry);
1440 	}
1441 
1442 	return result;
1443 }
1444 
1445 /*
1446  * Enqueues an event datagram to notify the peer VM attached to
1447  * the given queue pair handle about attach/detach event by the
1448  * given VM.  Returns Payload size of datagram enqueued on
1449  * success, error code otherwise.
1450  */
1451 static int qp_notify_peer(bool attach,
1452 			  struct vmci_handle handle,
1453 			  u32 my_id,
1454 			  u32 peer_id)
1455 {
1456 	int rv;
1457 	struct vmci_event_qp ev;
1458 
1459 	if (vmci_handle_is_invalid(handle) || my_id == VMCI_INVALID_ID ||
1460 	    peer_id == VMCI_INVALID_ID)
1461 		return VMCI_ERROR_INVALID_ARGS;
1462 
1463 	/*
1464 	 * In vmci_ctx_enqueue_datagram() we enforce the upper limit on
1465 	 * number of pending events from the hypervisor to a given VM
1466 	 * otherwise a rogue VM could do an arbitrary number of attach
1467 	 * and detach operations causing memory pressure in the host
1468 	 * kernel.
1469 	 */
1470 
1471 	memset(&ev, 0, sizeof(ev));
1472 	ev.msg.hdr.dst = vmci_make_handle(peer_id, VMCI_EVENT_HANDLER);
1473 	ev.msg.hdr.src = vmci_make_handle(VMCI_HYPERVISOR_CONTEXT_ID,
1474 					  VMCI_CONTEXT_RESOURCE_ID);
1475 	ev.msg.hdr.payload_size = sizeof(ev) - sizeof(ev.msg.hdr);
1476 	ev.msg.event_data.event = attach ?
1477 	    VMCI_EVENT_QP_PEER_ATTACH : VMCI_EVENT_QP_PEER_DETACH;
1478 	ev.payload.handle = handle;
1479 	ev.payload.peer_id = my_id;
1480 
1481 	rv = vmci_datagram_dispatch(VMCI_HYPERVISOR_CONTEXT_ID,
1482 				    &ev.msg.hdr, false);
1483 	if (rv < VMCI_SUCCESS)
1484 		pr_warn("Failed to enqueue queue_pair %s event datagram for context (ID=0x%x)\n",
1485 			attach ? "ATTACH" : "DETACH", peer_id);
1486 
1487 	return rv;
1488 }
1489 
1490 /*
1491  * The second endpoint issuing a queue pair allocation will attach to
1492  * the queue pair registered with the queue pair broker.
1493  *
1494  * If the attacher is a guest, it will associate a VMX virtual address
1495  * range with the queue pair as specified by the page_store. At this
1496  * point, the already attach host endpoint may start using the queue
1497  * pair, and an attach event is sent to it. For compatibility with
1498  * older VMX'en, that used a separate step to set the VMX virtual
1499  * address range, the virtual address range can be registered later
1500  * using vmci_qp_broker_set_page_store. In that case, a page_store of
1501  * NULL should be used, and the attach event will be generated once
1502  * the actual page store has been set.
1503  *
1504  * If the attacher is the host, a page_store of NULL should be used as
1505  * well, since the page store information is already set by the guest.
1506  *
1507  * For new VMX and host callers, the queue pair will be moved to the
1508  * VMCIQPB_ATTACHED_MEM state, and for older VMX callers, it will be
1509  * moved to the VMCOQPB_ATTACHED_NO_MEM state.
1510  */
1511 static int qp_broker_attach(struct qp_broker_entry *entry,
1512 			    u32 peer,
1513 			    u32 flags,
1514 			    u32 priv_flags,
1515 			    u64 produce_size,
1516 			    u64 consume_size,
1517 			    struct vmci_qp_page_store *page_store,
1518 			    struct vmci_ctx *context,
1519 			    vmci_event_release_cb wakeup_cb,
1520 			    void *client_data,
1521 			    struct qp_broker_entry **ent)
1522 {
1523 	const u32 context_id = vmci_ctx_get_id(context);
1524 	bool is_local = flags & VMCI_QPFLAG_LOCAL;
1525 	int result;
1526 
1527 	if (entry->state != VMCIQPB_CREATED_NO_MEM &&
1528 	    entry->state != VMCIQPB_CREATED_MEM)
1529 		return VMCI_ERROR_UNAVAILABLE;
1530 
1531 	if (is_local) {
1532 		if (!(entry->qp.flags & VMCI_QPFLAG_LOCAL) ||
1533 		    context_id != entry->create_id) {
1534 			return VMCI_ERROR_INVALID_ARGS;
1535 		}
1536 	} else if (context_id == entry->create_id ||
1537 		   context_id == entry->attach_id) {
1538 		return VMCI_ERROR_ALREADY_EXISTS;
1539 	}
1540 
1541 	if (VMCI_CONTEXT_IS_VM(context_id) &&
1542 	    VMCI_CONTEXT_IS_VM(entry->create_id))
1543 		return VMCI_ERROR_DST_UNREACHABLE;
1544 
1545 	/*
1546 	 * If we are attaching from a restricted context then the queuepair
1547 	 * must have been created by a trusted endpoint.
1548 	 */
1549 	if ((context->priv_flags & VMCI_PRIVILEGE_FLAG_RESTRICTED) &&
1550 	    !entry->created_by_trusted)
1551 		return VMCI_ERROR_NO_ACCESS;
1552 
1553 	/*
1554 	 * If we are attaching to a queuepair that was created by a restricted
1555 	 * context then we must be trusted.
1556 	 */
1557 	if (entry->require_trusted_attach &&
1558 	    (!(priv_flags & VMCI_PRIVILEGE_FLAG_TRUSTED)))
1559 		return VMCI_ERROR_NO_ACCESS;
1560 
1561 	/*
1562 	 * If the creator specifies VMCI_INVALID_ID in "peer" field, access
1563 	 * control check is not performed.
1564 	 */
1565 	if (entry->qp.peer != VMCI_INVALID_ID && entry->qp.peer != context_id)
1566 		return VMCI_ERROR_NO_ACCESS;
1567 
1568 	if (entry->create_id == VMCI_HOST_CONTEXT_ID) {
1569 		/*
1570 		 * Do not attach if the caller doesn't support Host Queue Pairs
1571 		 * and a host created this queue pair.
1572 		 */
1573 
1574 		if (!vmci_ctx_supports_host_qp(context))
1575 			return VMCI_ERROR_INVALID_RESOURCE;
1576 
1577 	} else if (context_id == VMCI_HOST_CONTEXT_ID) {
1578 		struct vmci_ctx *create_context;
1579 		bool supports_host_qp;
1580 
1581 		/*
1582 		 * Do not attach a host to a user created queue pair if that
1583 		 * user doesn't support host queue pair end points.
1584 		 */
1585 
1586 		create_context = vmci_ctx_get(entry->create_id);
1587 		supports_host_qp = vmci_ctx_supports_host_qp(create_context);
1588 		vmci_ctx_put(create_context);
1589 
1590 		if (!supports_host_qp)
1591 			return VMCI_ERROR_INVALID_RESOURCE;
1592 	}
1593 
1594 	if ((entry->qp.flags & ~VMCI_QP_ASYMM) != (flags & ~VMCI_QP_ASYMM_PEER))
1595 		return VMCI_ERROR_QUEUEPAIR_MISMATCH;
1596 
1597 	if (context_id != VMCI_HOST_CONTEXT_ID) {
1598 		/*
1599 		 * The queue pair broker entry stores values from the guest
1600 		 * point of view, so an attaching guest should match the values
1601 		 * stored in the entry.
1602 		 */
1603 
1604 		if (entry->qp.produce_size != produce_size ||
1605 		    entry->qp.consume_size != consume_size) {
1606 			return VMCI_ERROR_QUEUEPAIR_MISMATCH;
1607 		}
1608 	} else if (entry->qp.produce_size != consume_size ||
1609 		   entry->qp.consume_size != produce_size) {
1610 		return VMCI_ERROR_QUEUEPAIR_MISMATCH;
1611 	}
1612 
1613 	if (context_id != VMCI_HOST_CONTEXT_ID) {
1614 		/*
1615 		 * If a guest attached to a queue pair, it will supply
1616 		 * the backing memory.  If this is a pre NOVMVM vmx,
1617 		 * the backing memory will be supplied by calling
1618 		 * vmci_qp_broker_set_page_store() following the
1619 		 * return of the vmci_qp_broker_alloc() call. If it is
1620 		 * a vmx of version NOVMVM or later, the page store
1621 		 * must be supplied as part of the
1622 		 * vmci_qp_broker_alloc call.  Under all circumstances
1623 		 * must the initially created queue pair not have any
1624 		 * memory associated with it already.
1625 		 */
1626 
1627 		if (entry->state != VMCIQPB_CREATED_NO_MEM)
1628 			return VMCI_ERROR_INVALID_ARGS;
1629 
1630 		if (page_store != NULL) {
1631 			/*
1632 			 * Patch up host state to point to guest
1633 			 * supplied memory. The VMX already
1634 			 * initialized the queue pair headers, so no
1635 			 * need for the kernel side to do that.
1636 			 */
1637 
1638 			result = qp_host_register_user_memory(page_store,
1639 							      entry->produce_q,
1640 							      entry->consume_q);
1641 			if (result < VMCI_SUCCESS)
1642 				return result;
1643 
1644 			entry->state = VMCIQPB_ATTACHED_MEM;
1645 		} else {
1646 			entry->state = VMCIQPB_ATTACHED_NO_MEM;
1647 		}
1648 	} else if (entry->state == VMCIQPB_CREATED_NO_MEM) {
1649 		/*
1650 		 * The host side is attempting to attach to a queue
1651 		 * pair that doesn't have any memory associated with
1652 		 * it. This must be a pre NOVMVM vmx that hasn't set
1653 		 * the page store information yet, or a quiesced VM.
1654 		 */
1655 
1656 		return VMCI_ERROR_UNAVAILABLE;
1657 	} else {
1658 		/* The host side has successfully attached to a queue pair. */
1659 		entry->state = VMCIQPB_ATTACHED_MEM;
1660 	}
1661 
1662 	if (entry->state == VMCIQPB_ATTACHED_MEM) {
1663 		result =
1664 		    qp_notify_peer(true, entry->qp.handle, context_id,
1665 				   entry->create_id);
1666 		if (result < VMCI_SUCCESS)
1667 			pr_warn("Failed to notify peer (ID=0x%x) of attach to queue pair (handle=0x%x:0x%x)\n",
1668 				entry->create_id, entry->qp.handle.context,
1669 				entry->qp.handle.resource);
1670 	}
1671 
1672 	entry->attach_id = context_id;
1673 	entry->qp.ref_count++;
1674 	if (wakeup_cb) {
1675 		entry->wakeup_cb = wakeup_cb;
1676 		entry->client_data = client_data;
1677 	}
1678 
1679 	/*
1680 	 * When attaching to local queue pairs, the context already has
1681 	 * an entry tracking the queue pair, so don't add another one.
1682 	 */
1683 	if (!is_local)
1684 		vmci_ctx_qp_create(context, entry->qp.handle);
1685 
1686 	if (ent != NULL)
1687 		*ent = entry;
1688 
1689 	return VMCI_SUCCESS;
1690 }
1691 
1692 /*
1693  * queue_pair_Alloc for use when setting up queue pair endpoints
1694  * on the host.
1695  */
1696 static int qp_broker_alloc(struct vmci_handle handle,
1697 			   u32 peer,
1698 			   u32 flags,
1699 			   u32 priv_flags,
1700 			   u64 produce_size,
1701 			   u64 consume_size,
1702 			   struct vmci_qp_page_store *page_store,
1703 			   struct vmci_ctx *context,
1704 			   vmci_event_release_cb wakeup_cb,
1705 			   void *client_data,
1706 			   struct qp_broker_entry **ent,
1707 			   bool *swap)
1708 {
1709 	const u32 context_id = vmci_ctx_get_id(context);
1710 	bool create;
1711 	struct qp_broker_entry *entry = NULL;
1712 	bool is_local = flags & VMCI_QPFLAG_LOCAL;
1713 	int result;
1714 
1715 	if (vmci_handle_is_invalid(handle) ||
1716 	    (flags & ~VMCI_QP_ALL_FLAGS) || is_local ||
1717 	    !(produce_size || consume_size) ||
1718 	    !context || context_id == VMCI_INVALID_ID ||
1719 	    handle.context == VMCI_INVALID_ID) {
1720 		return VMCI_ERROR_INVALID_ARGS;
1721 	}
1722 
1723 	if (page_store && !VMCI_QP_PAGESTORE_IS_WELLFORMED(page_store))
1724 		return VMCI_ERROR_INVALID_ARGS;
1725 
1726 	/*
1727 	 * In the initial argument check, we ensure that non-vmkernel hosts
1728 	 * are not allowed to create local queue pairs.
1729 	 */
1730 
1731 	mutex_lock(&qp_broker_list.mutex);
1732 
1733 	if (!is_local && vmci_ctx_qp_exists(context, handle)) {
1734 		pr_devel("Context (ID=0x%x) already attached to queue pair (handle=0x%x:0x%x)\n",
1735 			 context_id, handle.context, handle.resource);
1736 		mutex_unlock(&qp_broker_list.mutex);
1737 		return VMCI_ERROR_ALREADY_EXISTS;
1738 	}
1739 
1740 	if (handle.resource != VMCI_INVALID_ID)
1741 		entry = qp_broker_handle_to_entry(handle);
1742 
1743 	if (!entry) {
1744 		create = true;
1745 		result =
1746 		    qp_broker_create(handle, peer, flags, priv_flags,
1747 				     produce_size, consume_size, page_store,
1748 				     context, wakeup_cb, client_data, ent);
1749 	} else {
1750 		create = false;
1751 		result =
1752 		    qp_broker_attach(entry, peer, flags, priv_flags,
1753 				     produce_size, consume_size, page_store,
1754 				     context, wakeup_cb, client_data, ent);
1755 	}
1756 
1757 	mutex_unlock(&qp_broker_list.mutex);
1758 
1759 	if (swap)
1760 		*swap = (context_id == VMCI_HOST_CONTEXT_ID) &&
1761 		    !(create && is_local);
1762 
1763 	return result;
1764 }
1765 
1766 /*
1767  * This function implements the kernel API for allocating a queue
1768  * pair.
1769  */
1770 static int qp_alloc_host_work(struct vmci_handle *handle,
1771 			      struct vmci_queue **produce_q,
1772 			      u64 produce_size,
1773 			      struct vmci_queue **consume_q,
1774 			      u64 consume_size,
1775 			      u32 peer,
1776 			      u32 flags,
1777 			      u32 priv_flags,
1778 			      vmci_event_release_cb wakeup_cb,
1779 			      void *client_data)
1780 {
1781 	struct vmci_handle new_handle;
1782 	struct vmci_ctx *context;
1783 	struct qp_broker_entry *entry;
1784 	int result;
1785 	bool swap;
1786 
1787 	if (vmci_handle_is_invalid(*handle)) {
1788 		new_handle = vmci_make_handle(
1789 			VMCI_HOST_CONTEXT_ID, VMCI_INVALID_ID);
1790 	} else
1791 		new_handle = *handle;
1792 
1793 	context = vmci_ctx_get(VMCI_HOST_CONTEXT_ID);
1794 	entry = NULL;
1795 	result =
1796 	    qp_broker_alloc(new_handle, peer, flags, priv_flags,
1797 			    produce_size, consume_size, NULL, context,
1798 			    wakeup_cb, client_data, &entry, &swap);
1799 	if (result == VMCI_SUCCESS) {
1800 		if (swap) {
1801 			/*
1802 			 * If this is a local queue pair, the attacher
1803 			 * will swap around produce and consume
1804 			 * queues.
1805 			 */
1806 
1807 			*produce_q = entry->consume_q;
1808 			*consume_q = entry->produce_q;
1809 		} else {
1810 			*produce_q = entry->produce_q;
1811 			*consume_q = entry->consume_q;
1812 		}
1813 
1814 		*handle = vmci_resource_handle(&entry->resource);
1815 	} else {
1816 		*handle = VMCI_INVALID_HANDLE;
1817 		pr_devel("queue pair broker failed to alloc (result=%d)\n",
1818 			 result);
1819 	}
1820 	vmci_ctx_put(context);
1821 	return result;
1822 }
1823 
1824 /*
1825  * Allocates a VMCI queue_pair. Only checks validity of input
1826  * arguments. The real work is done in the host or guest
1827  * specific function.
1828  */
1829 int vmci_qp_alloc(struct vmci_handle *handle,
1830 		  struct vmci_queue **produce_q,
1831 		  u64 produce_size,
1832 		  struct vmci_queue **consume_q,
1833 		  u64 consume_size,
1834 		  u32 peer,
1835 		  u32 flags,
1836 		  u32 priv_flags,
1837 		  bool guest_endpoint,
1838 		  vmci_event_release_cb wakeup_cb,
1839 		  void *client_data)
1840 {
1841 	if (!handle || !produce_q || !consume_q ||
1842 	    (!produce_size && !consume_size) || (flags & ~VMCI_QP_ALL_FLAGS))
1843 		return VMCI_ERROR_INVALID_ARGS;
1844 
1845 	if (guest_endpoint) {
1846 		return qp_alloc_guest_work(handle, produce_q,
1847 					   produce_size, consume_q,
1848 					   consume_size, peer,
1849 					   flags, priv_flags);
1850 	} else {
1851 		return qp_alloc_host_work(handle, produce_q,
1852 					  produce_size, consume_q,
1853 					  consume_size, peer, flags,
1854 					  priv_flags, wakeup_cb, client_data);
1855 	}
1856 }
1857 
1858 /*
1859  * This function implements the host kernel API for detaching from
1860  * a queue pair.
1861  */
1862 static int qp_detatch_host_work(struct vmci_handle handle)
1863 {
1864 	int result;
1865 	struct vmci_ctx *context;
1866 
1867 	context = vmci_ctx_get(VMCI_HOST_CONTEXT_ID);
1868 
1869 	result = vmci_qp_broker_detach(handle, context);
1870 
1871 	vmci_ctx_put(context);
1872 	return result;
1873 }
1874 
1875 /*
1876  * Detaches from a VMCI queue_pair. Only checks validity of input argument.
1877  * Real work is done in the host or guest specific function.
1878  */
1879 static int qp_detatch(struct vmci_handle handle, bool guest_endpoint)
1880 {
1881 	if (vmci_handle_is_invalid(handle))
1882 		return VMCI_ERROR_INVALID_ARGS;
1883 
1884 	if (guest_endpoint)
1885 		return qp_detatch_guest_work(handle);
1886 	else
1887 		return qp_detatch_host_work(handle);
1888 }
1889 
1890 /*
1891  * Returns the entry from the head of the list. Assumes that the list is
1892  * locked.
1893  */
1894 static struct qp_entry *qp_list_get_head(struct qp_list *qp_list)
1895 {
1896 	if (!list_empty(&qp_list->head)) {
1897 		struct qp_entry *entry =
1898 		    list_first_entry(&qp_list->head, struct qp_entry,
1899 				     list_item);
1900 		return entry;
1901 	}
1902 
1903 	return NULL;
1904 }
1905 
1906 void vmci_qp_broker_exit(void)
1907 {
1908 	struct qp_entry *entry;
1909 	struct qp_broker_entry *be;
1910 
1911 	mutex_lock(&qp_broker_list.mutex);
1912 
1913 	while ((entry = qp_list_get_head(&qp_broker_list))) {
1914 		be = (struct qp_broker_entry *)entry;
1915 
1916 		qp_list_remove_entry(&qp_broker_list, entry);
1917 		kfree(be);
1918 	}
1919 
1920 	mutex_unlock(&qp_broker_list.mutex);
1921 }
1922 
1923 /*
1924  * Requests that a queue pair be allocated with the VMCI queue
1925  * pair broker. Allocates a queue pair entry if one does not
1926  * exist. Attaches to one if it exists, and retrieves the page
1927  * files backing that queue_pair.  Assumes that the queue pair
1928  * broker lock is held.
1929  */
1930 int vmci_qp_broker_alloc(struct vmci_handle handle,
1931 			 u32 peer,
1932 			 u32 flags,
1933 			 u32 priv_flags,
1934 			 u64 produce_size,
1935 			 u64 consume_size,
1936 			 struct vmci_qp_page_store *page_store,
1937 			 struct vmci_ctx *context)
1938 {
1939 	if (!QP_SIZES_ARE_VALID(produce_size, consume_size))
1940 		return VMCI_ERROR_NO_RESOURCES;
1941 
1942 	return qp_broker_alloc(handle, peer, flags, priv_flags,
1943 			       produce_size, consume_size,
1944 			       page_store, context, NULL, NULL, NULL, NULL);
1945 }
1946 
1947 /*
1948  * VMX'en with versions lower than VMCI_VERSION_NOVMVM use a separate
1949  * step to add the UVAs of the VMX mapping of the queue pair. This function
1950  * provides backwards compatibility with such VMX'en, and takes care of
1951  * registering the page store for a queue pair previously allocated by the
1952  * VMX during create or attach. This function will move the queue pair state
1953  * to either from VMCIQBP_CREATED_NO_MEM to VMCIQBP_CREATED_MEM or
1954  * VMCIQBP_ATTACHED_NO_MEM to VMCIQBP_ATTACHED_MEM. If moving to the
1955  * attached state with memory, the queue pair is ready to be used by the
1956  * host peer, and an attached event will be generated.
1957  *
1958  * Assumes that the queue pair broker lock is held.
1959  *
1960  * This function is only used by the hosted platform, since there is no
1961  * issue with backwards compatibility for vmkernel.
1962  */
1963 int vmci_qp_broker_set_page_store(struct vmci_handle handle,
1964 				  u64 produce_uva,
1965 				  u64 consume_uva,
1966 				  struct vmci_ctx *context)
1967 {
1968 	struct qp_broker_entry *entry;
1969 	int result;
1970 	const u32 context_id = vmci_ctx_get_id(context);
1971 
1972 	if (vmci_handle_is_invalid(handle) || !context ||
1973 	    context_id == VMCI_INVALID_ID)
1974 		return VMCI_ERROR_INVALID_ARGS;
1975 
1976 	/*
1977 	 * We only support guest to host queue pairs, so the VMX must
1978 	 * supply UVAs for the mapped page files.
1979 	 */
1980 
1981 	if (produce_uva == 0 || consume_uva == 0)
1982 		return VMCI_ERROR_INVALID_ARGS;
1983 
1984 	mutex_lock(&qp_broker_list.mutex);
1985 
1986 	if (!vmci_ctx_qp_exists(context, handle)) {
1987 		pr_warn("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
1988 			context_id, handle.context, handle.resource);
1989 		result = VMCI_ERROR_NOT_FOUND;
1990 		goto out;
1991 	}
1992 
1993 	entry = qp_broker_handle_to_entry(handle);
1994 	if (!entry) {
1995 		result = VMCI_ERROR_NOT_FOUND;
1996 		goto out;
1997 	}
1998 
1999 	/*
2000 	 * If I'm the owner then I can set the page store.
2001 	 *
2002 	 * Or, if a host created the queue_pair and I'm the attached peer
2003 	 * then I can set the page store.
2004 	 */
2005 	if (entry->create_id != context_id &&
2006 	    (entry->create_id != VMCI_HOST_CONTEXT_ID ||
2007 	     entry->attach_id != context_id)) {
2008 		result = VMCI_ERROR_QUEUEPAIR_NOTOWNER;
2009 		goto out;
2010 	}
2011 
2012 	if (entry->state != VMCIQPB_CREATED_NO_MEM &&
2013 	    entry->state != VMCIQPB_ATTACHED_NO_MEM) {
2014 		result = VMCI_ERROR_UNAVAILABLE;
2015 		goto out;
2016 	}
2017 
2018 	result = qp_host_get_user_memory(produce_uva, consume_uva,
2019 					 entry->produce_q, entry->consume_q);
2020 	if (result < VMCI_SUCCESS)
2021 		goto out;
2022 
2023 	result = qp_host_map_queues(entry->produce_q, entry->consume_q);
2024 	if (result < VMCI_SUCCESS) {
2025 		qp_host_unregister_user_memory(entry->produce_q,
2026 					       entry->consume_q);
2027 		goto out;
2028 	}
2029 
2030 	if (entry->state == VMCIQPB_CREATED_NO_MEM)
2031 		entry->state = VMCIQPB_CREATED_MEM;
2032 	else
2033 		entry->state = VMCIQPB_ATTACHED_MEM;
2034 
2035 	entry->vmci_page_files = true;
2036 
2037 	if (entry->state == VMCIQPB_ATTACHED_MEM) {
2038 		result =
2039 		    qp_notify_peer(true, handle, context_id, entry->create_id);
2040 		if (result < VMCI_SUCCESS) {
2041 			pr_warn("Failed to notify peer (ID=0x%x) of attach to queue pair (handle=0x%x:0x%x)\n",
2042 				entry->create_id, entry->qp.handle.context,
2043 				entry->qp.handle.resource);
2044 		}
2045 	}
2046 
2047 	result = VMCI_SUCCESS;
2048  out:
2049 	mutex_unlock(&qp_broker_list.mutex);
2050 	return result;
2051 }
2052 
2053 /*
2054  * Resets saved queue headers for the given QP broker
2055  * entry. Should be used when guest memory becomes available
2056  * again, or the guest detaches.
2057  */
2058 static void qp_reset_saved_headers(struct qp_broker_entry *entry)
2059 {
2060 	entry->produce_q->saved_header = NULL;
2061 	entry->consume_q->saved_header = NULL;
2062 }
2063 
2064 /*
2065  * The main entry point for detaching from a queue pair registered with the
2066  * queue pair broker. If more than one endpoint is attached to the queue
2067  * pair, the first endpoint will mainly decrement a reference count and
2068  * generate a notification to its peer. The last endpoint will clean up
2069  * the queue pair state registered with the broker.
2070  *
2071  * When a guest endpoint detaches, it will unmap and unregister the guest
2072  * memory backing the queue pair. If the host is still attached, it will
2073  * no longer be able to access the queue pair content.
2074  *
2075  * If the queue pair is already in a state where there is no memory
2076  * registered for the queue pair (any *_NO_MEM state), it will transition to
2077  * the VMCIQPB_SHUTDOWN_NO_MEM state. This will also happen, if a guest
2078  * endpoint is the first of two endpoints to detach. If the host endpoint is
2079  * the first out of two to detach, the queue pair will move to the
2080  * VMCIQPB_SHUTDOWN_MEM state.
2081  */
2082 int vmci_qp_broker_detach(struct vmci_handle handle, struct vmci_ctx *context)
2083 {
2084 	struct qp_broker_entry *entry;
2085 	const u32 context_id = vmci_ctx_get_id(context);
2086 	u32 peer_id;
2087 	bool is_local = false;
2088 	int result;
2089 
2090 	if (vmci_handle_is_invalid(handle) || !context ||
2091 	    context_id == VMCI_INVALID_ID) {
2092 		return VMCI_ERROR_INVALID_ARGS;
2093 	}
2094 
2095 	mutex_lock(&qp_broker_list.mutex);
2096 
2097 	if (!vmci_ctx_qp_exists(context, handle)) {
2098 		pr_devel("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
2099 			 context_id, handle.context, handle.resource);
2100 		result = VMCI_ERROR_NOT_FOUND;
2101 		goto out;
2102 	}
2103 
2104 	entry = qp_broker_handle_to_entry(handle);
2105 	if (!entry) {
2106 		pr_devel("Context (ID=0x%x) reports being attached to queue pair(handle=0x%x:0x%x) that isn't present in broker\n",
2107 			 context_id, handle.context, handle.resource);
2108 		result = VMCI_ERROR_NOT_FOUND;
2109 		goto out;
2110 	}
2111 
2112 	if (context_id != entry->create_id && context_id != entry->attach_id) {
2113 		result = VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2114 		goto out;
2115 	}
2116 
2117 	if (context_id == entry->create_id) {
2118 		peer_id = entry->attach_id;
2119 		entry->create_id = VMCI_INVALID_ID;
2120 	} else {
2121 		peer_id = entry->create_id;
2122 		entry->attach_id = VMCI_INVALID_ID;
2123 	}
2124 	entry->qp.ref_count--;
2125 
2126 	is_local = entry->qp.flags & VMCI_QPFLAG_LOCAL;
2127 
2128 	if (context_id != VMCI_HOST_CONTEXT_ID) {
2129 		bool headers_mapped;
2130 
2131 		/*
2132 		 * Pre NOVMVM vmx'en may detach from a queue pair
2133 		 * before setting the page store, and in that case
2134 		 * there is no user memory to detach from. Also, more
2135 		 * recent VMX'en may detach from a queue pair in the
2136 		 * quiesced state.
2137 		 */
2138 
2139 		qp_acquire_queue_mutex(entry->produce_q);
2140 		headers_mapped = entry->produce_q->q_header ||
2141 		    entry->consume_q->q_header;
2142 		if (QPBROKERSTATE_HAS_MEM(entry)) {
2143 			result =
2144 			    qp_host_unmap_queues(INVALID_VMCI_GUEST_MEM_ID,
2145 						 entry->produce_q,
2146 						 entry->consume_q);
2147 			if (result < VMCI_SUCCESS)
2148 				pr_warn("Failed to unmap queue headers for queue pair (handle=0x%x:0x%x,result=%d)\n",
2149 					handle.context, handle.resource,
2150 					result);
2151 
2152 			qp_host_unregister_user_memory(entry->produce_q,
2153 						       entry->consume_q);
2154 
2155 		}
2156 
2157 		if (!headers_mapped)
2158 			qp_reset_saved_headers(entry);
2159 
2160 		qp_release_queue_mutex(entry->produce_q);
2161 
2162 		if (!headers_mapped && entry->wakeup_cb)
2163 			entry->wakeup_cb(entry->client_data);
2164 
2165 	} else {
2166 		if (entry->wakeup_cb) {
2167 			entry->wakeup_cb = NULL;
2168 			entry->client_data = NULL;
2169 		}
2170 	}
2171 
2172 	if (entry->qp.ref_count == 0) {
2173 		qp_list_remove_entry(&qp_broker_list, &entry->qp);
2174 
2175 		if (is_local)
2176 			kfree(entry->local_mem);
2177 
2178 		qp_cleanup_queue_mutex(entry->produce_q, entry->consume_q);
2179 		qp_host_free_queue(entry->produce_q, entry->qp.produce_size);
2180 		qp_host_free_queue(entry->consume_q, entry->qp.consume_size);
2181 		/* Unlink from resource hash table and free callback */
2182 		vmci_resource_remove(&entry->resource);
2183 
2184 		kfree(entry);
2185 
2186 		vmci_ctx_qp_destroy(context, handle);
2187 	} else {
2188 		qp_notify_peer(false, handle, context_id, peer_id);
2189 		if (context_id == VMCI_HOST_CONTEXT_ID &&
2190 		    QPBROKERSTATE_HAS_MEM(entry)) {
2191 			entry->state = VMCIQPB_SHUTDOWN_MEM;
2192 		} else {
2193 			entry->state = VMCIQPB_SHUTDOWN_NO_MEM;
2194 		}
2195 
2196 		if (!is_local)
2197 			vmci_ctx_qp_destroy(context, handle);
2198 
2199 	}
2200 	result = VMCI_SUCCESS;
2201  out:
2202 	mutex_unlock(&qp_broker_list.mutex);
2203 	return result;
2204 }
2205 
2206 /*
2207  * Establishes the necessary mappings for a queue pair given a
2208  * reference to the queue pair guest memory. This is usually
2209  * called when a guest is unquiesced and the VMX is allowed to
2210  * map guest memory once again.
2211  */
2212 int vmci_qp_broker_map(struct vmci_handle handle,
2213 		       struct vmci_ctx *context,
2214 		       u64 guest_mem)
2215 {
2216 	struct qp_broker_entry *entry;
2217 	const u32 context_id = vmci_ctx_get_id(context);
2218 	int result;
2219 
2220 	if (vmci_handle_is_invalid(handle) || !context ||
2221 	    context_id == VMCI_INVALID_ID)
2222 		return VMCI_ERROR_INVALID_ARGS;
2223 
2224 	mutex_lock(&qp_broker_list.mutex);
2225 
2226 	if (!vmci_ctx_qp_exists(context, handle)) {
2227 		pr_devel("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
2228 			 context_id, handle.context, handle.resource);
2229 		result = VMCI_ERROR_NOT_FOUND;
2230 		goto out;
2231 	}
2232 
2233 	entry = qp_broker_handle_to_entry(handle);
2234 	if (!entry) {
2235 		pr_devel("Context (ID=0x%x) reports being attached to queue pair (handle=0x%x:0x%x) that isn't present in broker\n",
2236 			 context_id, handle.context, handle.resource);
2237 		result = VMCI_ERROR_NOT_FOUND;
2238 		goto out;
2239 	}
2240 
2241 	if (context_id != entry->create_id && context_id != entry->attach_id) {
2242 		result = VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2243 		goto out;
2244 	}
2245 
2246 	result = VMCI_SUCCESS;
2247 
2248 	if (context_id != VMCI_HOST_CONTEXT_ID &&
2249 	    !QPBROKERSTATE_HAS_MEM(entry)) {
2250 		struct vmci_qp_page_store page_store;
2251 
2252 		page_store.pages = guest_mem;
2253 		page_store.len = QPE_NUM_PAGES(entry->qp);
2254 
2255 		qp_acquire_queue_mutex(entry->produce_q);
2256 		qp_reset_saved_headers(entry);
2257 		result =
2258 		    qp_host_register_user_memory(&page_store,
2259 						 entry->produce_q,
2260 						 entry->consume_q);
2261 		qp_release_queue_mutex(entry->produce_q);
2262 		if (result == VMCI_SUCCESS) {
2263 			/* Move state from *_NO_MEM to *_MEM */
2264 
2265 			entry->state++;
2266 
2267 			if (entry->wakeup_cb)
2268 				entry->wakeup_cb(entry->client_data);
2269 		}
2270 	}
2271 
2272  out:
2273 	mutex_unlock(&qp_broker_list.mutex);
2274 	return result;
2275 }
2276 
2277 /*
2278  * Saves a snapshot of the queue headers for the given QP broker
2279  * entry. Should be used when guest memory is unmapped.
2280  * Results:
2281  * VMCI_SUCCESS on success, appropriate error code if guest memory
2282  * can't be accessed..
2283  */
2284 static int qp_save_headers(struct qp_broker_entry *entry)
2285 {
2286 	int result;
2287 
2288 	if (entry->produce_q->saved_header != NULL &&
2289 	    entry->consume_q->saved_header != NULL) {
2290 		/*
2291 		 *  If the headers have already been saved, we don't need to do
2292 		 *  it again, and we don't want to map in the headers
2293 		 *  unnecessarily.
2294 		 */
2295 
2296 		return VMCI_SUCCESS;
2297 	}
2298 
2299 	if (NULL == entry->produce_q->q_header ||
2300 	    NULL == entry->consume_q->q_header) {
2301 		result = qp_host_map_queues(entry->produce_q, entry->consume_q);
2302 		if (result < VMCI_SUCCESS)
2303 			return result;
2304 	}
2305 
2306 	memcpy(&entry->saved_produce_q, entry->produce_q->q_header,
2307 	       sizeof(entry->saved_produce_q));
2308 	entry->produce_q->saved_header = &entry->saved_produce_q;
2309 	memcpy(&entry->saved_consume_q, entry->consume_q->q_header,
2310 	       sizeof(entry->saved_consume_q));
2311 	entry->consume_q->saved_header = &entry->saved_consume_q;
2312 
2313 	return VMCI_SUCCESS;
2314 }
2315 
2316 /*
2317  * Removes all references to the guest memory of a given queue pair, and
2318  * will move the queue pair from state *_MEM to *_NO_MEM. It is usually
2319  * called when a VM is being quiesced where access to guest memory should
2320  * avoided.
2321  */
2322 int vmci_qp_broker_unmap(struct vmci_handle handle,
2323 			 struct vmci_ctx *context,
2324 			 u32 gid)
2325 {
2326 	struct qp_broker_entry *entry;
2327 	const u32 context_id = vmci_ctx_get_id(context);
2328 	int result;
2329 
2330 	if (vmci_handle_is_invalid(handle) || !context ||
2331 	    context_id == VMCI_INVALID_ID)
2332 		return VMCI_ERROR_INVALID_ARGS;
2333 
2334 	mutex_lock(&qp_broker_list.mutex);
2335 
2336 	if (!vmci_ctx_qp_exists(context, handle)) {
2337 		pr_devel("Context (ID=0x%x) not attached to queue pair (handle=0x%x:0x%x)\n",
2338 			 context_id, handle.context, handle.resource);
2339 		result = VMCI_ERROR_NOT_FOUND;
2340 		goto out;
2341 	}
2342 
2343 	entry = qp_broker_handle_to_entry(handle);
2344 	if (!entry) {
2345 		pr_devel("Context (ID=0x%x) reports being attached to queue pair (handle=0x%x:0x%x) that isn't present in broker\n",
2346 			 context_id, handle.context, handle.resource);
2347 		result = VMCI_ERROR_NOT_FOUND;
2348 		goto out;
2349 	}
2350 
2351 	if (context_id != entry->create_id && context_id != entry->attach_id) {
2352 		result = VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2353 		goto out;
2354 	}
2355 
2356 	if (context_id != VMCI_HOST_CONTEXT_ID &&
2357 	    QPBROKERSTATE_HAS_MEM(entry)) {
2358 		qp_acquire_queue_mutex(entry->produce_q);
2359 		result = qp_save_headers(entry);
2360 		if (result < VMCI_SUCCESS)
2361 			pr_warn("Failed to save queue headers for queue pair (handle=0x%x:0x%x,result=%d)\n",
2362 				handle.context, handle.resource, result);
2363 
2364 		qp_host_unmap_queues(gid, entry->produce_q, entry->consume_q);
2365 
2366 		/*
2367 		 * On hosted, when we unmap queue pairs, the VMX will also
2368 		 * unmap the guest memory, so we invalidate the previously
2369 		 * registered memory. If the queue pair is mapped again at a
2370 		 * later point in time, we will need to reregister the user
2371 		 * memory with a possibly new user VA.
2372 		 */
2373 		qp_host_unregister_user_memory(entry->produce_q,
2374 					       entry->consume_q);
2375 
2376 		/*
2377 		 * Move state from *_MEM to *_NO_MEM.
2378 		 */
2379 		entry->state--;
2380 
2381 		qp_release_queue_mutex(entry->produce_q);
2382 	}
2383 
2384 	result = VMCI_SUCCESS;
2385 
2386  out:
2387 	mutex_unlock(&qp_broker_list.mutex);
2388 	return result;
2389 }
2390 
2391 /*
2392  * Destroys all guest queue pair endpoints. If active guest queue
2393  * pairs still exist, hypercalls to attempt detach from these
2394  * queue pairs will be made. Any failure to detach is silently
2395  * ignored.
2396  */
2397 void vmci_qp_guest_endpoints_exit(void)
2398 {
2399 	struct qp_entry *entry;
2400 	struct qp_guest_endpoint *ep;
2401 
2402 	mutex_lock(&qp_guest_endpoints.mutex);
2403 
2404 	while ((entry = qp_list_get_head(&qp_guest_endpoints))) {
2405 		ep = (struct qp_guest_endpoint *)entry;
2406 
2407 		/* Don't make a hypercall for local queue_pairs. */
2408 		if (!(entry->flags & VMCI_QPFLAG_LOCAL))
2409 			qp_detatch_hypercall(entry->handle);
2410 
2411 		/* We cannot fail the exit, so let's reset ref_count. */
2412 		entry->ref_count = 0;
2413 		qp_list_remove_entry(&qp_guest_endpoints, entry);
2414 
2415 		qp_guest_endpoint_destroy(ep);
2416 	}
2417 
2418 	mutex_unlock(&qp_guest_endpoints.mutex);
2419 }
2420 
2421 /*
2422  * Helper routine that will lock the queue pair before subsequent
2423  * operations.
2424  * Note: Non-blocking on the host side is currently only implemented in ESX.
2425  * Since non-blocking isn't yet implemented on the host personality we
2426  * have no reason to acquire a spin lock.  So to avoid the use of an
2427  * unnecessary lock only acquire the mutex if we can block.
2428  */
2429 static void qp_lock(const struct vmci_qp *qpair)
2430 {
2431 	qp_acquire_queue_mutex(qpair->produce_q);
2432 }
2433 
2434 /*
2435  * Helper routine that unlocks the queue pair after calling
2436  * qp_lock.
2437  */
2438 static void qp_unlock(const struct vmci_qp *qpair)
2439 {
2440 	qp_release_queue_mutex(qpair->produce_q);
2441 }
2442 
2443 /*
2444  * The queue headers may not be mapped at all times. If a queue is
2445  * currently not mapped, it will be attempted to do so.
2446  */
2447 static int qp_map_queue_headers(struct vmci_queue *produce_q,
2448 				struct vmci_queue *consume_q)
2449 {
2450 	int result;
2451 
2452 	if (NULL == produce_q->q_header || NULL == consume_q->q_header) {
2453 		result = qp_host_map_queues(produce_q, consume_q);
2454 		if (result < VMCI_SUCCESS)
2455 			return (produce_q->saved_header &&
2456 				consume_q->saved_header) ?
2457 			    VMCI_ERROR_QUEUEPAIR_NOT_READY :
2458 			    VMCI_ERROR_QUEUEPAIR_NOTATTACHED;
2459 	}
2460 
2461 	return VMCI_SUCCESS;
2462 }
2463 
2464 /*
2465  * Helper routine that will retrieve the produce and consume
2466  * headers of a given queue pair. If the guest memory of the
2467  * queue pair is currently not available, the saved queue headers
2468  * will be returned, if these are available.
2469  */
2470 static int qp_get_queue_headers(const struct vmci_qp *qpair,
2471 				struct vmci_queue_header **produce_q_header,
2472 				struct vmci_queue_header **consume_q_header)
2473 {
2474 	int result;
2475 
2476 	result = qp_map_queue_headers(qpair->produce_q, qpair->consume_q);
2477 	if (result == VMCI_SUCCESS) {
2478 		*produce_q_header = qpair->produce_q->q_header;
2479 		*consume_q_header = qpair->consume_q->q_header;
2480 	} else if (qpair->produce_q->saved_header &&
2481 		   qpair->consume_q->saved_header) {
2482 		*produce_q_header = qpair->produce_q->saved_header;
2483 		*consume_q_header = qpair->consume_q->saved_header;
2484 		result = VMCI_SUCCESS;
2485 	}
2486 
2487 	return result;
2488 }
2489 
2490 /*
2491  * Callback from VMCI queue pair broker indicating that a queue
2492  * pair that was previously not ready, now either is ready or
2493  * gone forever.
2494  */
2495 static int qp_wakeup_cb(void *client_data)
2496 {
2497 	struct vmci_qp *qpair = (struct vmci_qp *)client_data;
2498 
2499 	qp_lock(qpair);
2500 	while (qpair->blocked > 0) {
2501 		qpair->blocked--;
2502 		qpair->generation++;
2503 		wake_up(&qpair->event);
2504 	}
2505 	qp_unlock(qpair);
2506 
2507 	return VMCI_SUCCESS;
2508 }
2509 
2510 /*
2511  * Makes the calling thread wait for the queue pair to become
2512  * ready for host side access.  Returns true when thread is
2513  * woken up after queue pair state change, false otherwise.
2514  */
2515 static bool qp_wait_for_ready_queue(struct vmci_qp *qpair)
2516 {
2517 	unsigned int generation;
2518 
2519 	qpair->blocked++;
2520 	generation = qpair->generation;
2521 	qp_unlock(qpair);
2522 	wait_event(qpair->event, generation != qpair->generation);
2523 	qp_lock(qpair);
2524 
2525 	return true;
2526 }
2527 
2528 /*
2529  * Enqueues a given buffer to the produce queue using the provided
2530  * function. As many bytes as possible (space available in the queue)
2531  * are enqueued.  Assumes the queue->mutex has been acquired.  Returns
2532  * VMCI_ERROR_QUEUEPAIR_NOSPACE if no space was available to enqueue
2533  * data, VMCI_ERROR_INVALID_SIZE, if any queue pointer is outside the
2534  * queue (as defined by the queue size), VMCI_ERROR_INVALID_ARGS, if
2535  * an error occured when accessing the buffer,
2536  * VMCI_ERROR_QUEUEPAIR_NOTATTACHED, if the queue pair pages aren't
2537  * available.  Otherwise, the number of bytes written to the queue is
2538  * returned.  Updates the tail pointer of the produce queue.
2539  */
2540 static ssize_t qp_enqueue_locked(struct vmci_queue *produce_q,
2541 				 struct vmci_queue *consume_q,
2542 				 const u64 produce_q_size,
2543 				 struct iov_iter *from)
2544 {
2545 	s64 free_space;
2546 	u64 tail;
2547 	size_t buf_size = iov_iter_count(from);
2548 	size_t written;
2549 	ssize_t result;
2550 
2551 	result = qp_map_queue_headers(produce_q, consume_q);
2552 	if (unlikely(result != VMCI_SUCCESS))
2553 		return result;
2554 
2555 	free_space = vmci_q_header_free_space(produce_q->q_header,
2556 					      consume_q->q_header,
2557 					      produce_q_size);
2558 	if (free_space == 0)
2559 		return VMCI_ERROR_QUEUEPAIR_NOSPACE;
2560 
2561 	if (free_space < VMCI_SUCCESS)
2562 		return (ssize_t) free_space;
2563 
2564 	written = (size_t) (free_space > buf_size ? buf_size : free_space);
2565 	tail = vmci_q_header_producer_tail(produce_q->q_header);
2566 	if (likely(tail + written < produce_q_size)) {
2567 		result = qp_memcpy_to_queue_iter(produce_q, tail, from, written);
2568 	} else {
2569 		/* Tail pointer wraps around. */
2570 
2571 		const size_t tmp = (size_t) (produce_q_size - tail);
2572 
2573 		result = qp_memcpy_to_queue_iter(produce_q, tail, from, tmp);
2574 		if (result >= VMCI_SUCCESS)
2575 			result = qp_memcpy_to_queue_iter(produce_q, 0, from,
2576 						 written - tmp);
2577 	}
2578 
2579 	if (result < VMCI_SUCCESS)
2580 		return result;
2581 
2582 	/*
2583 	 * This virt_wmb() ensures that data written to the queue
2584 	 * is observable before the new producer_tail is.
2585 	 */
2586 	virt_wmb();
2587 
2588 	vmci_q_header_add_producer_tail(produce_q->q_header, written,
2589 					produce_q_size);
2590 	return written;
2591 }
2592 
2593 /*
2594  * Dequeues data (if available) from the given consume queue. Writes data
2595  * to the user provided buffer using the provided function.
2596  * Assumes the queue->mutex has been acquired.
2597  * Results:
2598  * VMCI_ERROR_QUEUEPAIR_NODATA if no data was available to dequeue.
2599  * VMCI_ERROR_INVALID_SIZE, if any queue pointer is outside the queue
2600  * (as defined by the queue size).
2601  * VMCI_ERROR_INVALID_ARGS, if an error occured when accessing the buffer.
2602  * Otherwise the number of bytes dequeued is returned.
2603  * Side effects:
2604  * Updates the head pointer of the consume queue.
2605  */
2606 static ssize_t qp_dequeue_locked(struct vmci_queue *produce_q,
2607 				 struct vmci_queue *consume_q,
2608 				 const u64 consume_q_size,
2609 				 struct iov_iter *to,
2610 				 bool update_consumer)
2611 {
2612 	size_t buf_size = iov_iter_count(to);
2613 	s64 buf_ready;
2614 	u64 head;
2615 	size_t read;
2616 	ssize_t result;
2617 
2618 	result = qp_map_queue_headers(produce_q, consume_q);
2619 	if (unlikely(result != VMCI_SUCCESS))
2620 		return result;
2621 
2622 	buf_ready = vmci_q_header_buf_ready(consume_q->q_header,
2623 					    produce_q->q_header,
2624 					    consume_q_size);
2625 	if (buf_ready == 0)
2626 		return VMCI_ERROR_QUEUEPAIR_NODATA;
2627 
2628 	if (buf_ready < VMCI_SUCCESS)
2629 		return (ssize_t) buf_ready;
2630 
2631 	/*
2632 	 * This virt_rmb() ensures that data from the queue will be read
2633 	 * after we have determined how much is ready to be consumed.
2634 	 */
2635 	virt_rmb();
2636 
2637 	read = (size_t) (buf_ready > buf_size ? buf_size : buf_ready);
2638 	head = vmci_q_header_consumer_head(produce_q->q_header);
2639 	if (likely(head + read < consume_q_size)) {
2640 		result = qp_memcpy_from_queue_iter(to, consume_q, head, read);
2641 	} else {
2642 		/* Head pointer wraps around. */
2643 
2644 		const size_t tmp = (size_t) (consume_q_size - head);
2645 
2646 		result = qp_memcpy_from_queue_iter(to, consume_q, head, tmp);
2647 		if (result >= VMCI_SUCCESS)
2648 			result = qp_memcpy_from_queue_iter(to, consume_q, 0,
2649 						   read - tmp);
2650 
2651 	}
2652 
2653 	if (result < VMCI_SUCCESS)
2654 		return result;
2655 
2656 	if (update_consumer)
2657 		vmci_q_header_add_consumer_head(produce_q->q_header,
2658 						read, consume_q_size);
2659 
2660 	return read;
2661 }
2662 
2663 /*
2664  * vmci_qpair_alloc() - Allocates a queue pair.
2665  * @qpair:      Pointer for the new vmci_qp struct.
2666  * @handle:     Handle to track the resource.
2667  * @produce_qsize:      Desired size of the producer queue.
2668  * @consume_qsize:      Desired size of the consumer queue.
2669  * @peer:       ContextID of the peer.
2670  * @flags:      VMCI flags.
2671  * @priv_flags: VMCI priviledge flags.
2672  *
2673  * This is the client interface for allocating the memory for a
2674  * vmci_qp structure and then attaching to the underlying
2675  * queue.  If an error occurs allocating the memory for the
2676  * vmci_qp structure no attempt is made to attach.  If an
2677  * error occurs attaching, then the structure is freed.
2678  */
2679 int vmci_qpair_alloc(struct vmci_qp **qpair,
2680 		     struct vmci_handle *handle,
2681 		     u64 produce_qsize,
2682 		     u64 consume_qsize,
2683 		     u32 peer,
2684 		     u32 flags,
2685 		     u32 priv_flags)
2686 {
2687 	struct vmci_qp *my_qpair;
2688 	int retval;
2689 	struct vmci_handle src = VMCI_INVALID_HANDLE;
2690 	struct vmci_handle dst = vmci_make_handle(peer, VMCI_INVALID_ID);
2691 	enum vmci_route route;
2692 	vmci_event_release_cb wakeup_cb;
2693 	void *client_data;
2694 
2695 	/*
2696 	 * Restrict the size of a queuepair.  The device already
2697 	 * enforces a limit on the total amount of memory that can be
2698 	 * allocated to queuepairs for a guest.  However, we try to
2699 	 * allocate this memory before we make the queuepair
2700 	 * allocation hypercall.  On Linux, we allocate each page
2701 	 * separately, which means rather than fail, the guest will
2702 	 * thrash while it tries to allocate, and will become
2703 	 * increasingly unresponsive to the point where it appears to
2704 	 * be hung.  So we place a limit on the size of an individual
2705 	 * queuepair here, and leave the device to enforce the
2706 	 * restriction on total queuepair memory.  (Note that this
2707 	 * doesn't prevent all cases; a user with only this much
2708 	 * physical memory could still get into trouble.)  The error
2709 	 * used by the device is NO_RESOURCES, so use that here too.
2710 	 */
2711 
2712 	if (!QP_SIZES_ARE_VALID(produce_qsize, consume_qsize))
2713 		return VMCI_ERROR_NO_RESOURCES;
2714 
2715 	retval = vmci_route(&src, &dst, false, &route);
2716 	if (retval < VMCI_SUCCESS)
2717 		route = vmci_guest_code_active() ?
2718 		    VMCI_ROUTE_AS_GUEST : VMCI_ROUTE_AS_HOST;
2719 
2720 	if (flags & (VMCI_QPFLAG_NONBLOCK | VMCI_QPFLAG_PINNED)) {
2721 		pr_devel("NONBLOCK OR PINNED set");
2722 		return VMCI_ERROR_INVALID_ARGS;
2723 	}
2724 
2725 	my_qpair = kzalloc(sizeof(*my_qpair), GFP_KERNEL);
2726 	if (!my_qpair)
2727 		return VMCI_ERROR_NO_MEM;
2728 
2729 	my_qpair->produce_q_size = produce_qsize;
2730 	my_qpair->consume_q_size = consume_qsize;
2731 	my_qpair->peer = peer;
2732 	my_qpair->flags = flags;
2733 	my_qpair->priv_flags = priv_flags;
2734 
2735 	wakeup_cb = NULL;
2736 	client_data = NULL;
2737 
2738 	if (VMCI_ROUTE_AS_HOST == route) {
2739 		my_qpair->guest_endpoint = false;
2740 		if (!(flags & VMCI_QPFLAG_LOCAL)) {
2741 			my_qpair->blocked = 0;
2742 			my_qpair->generation = 0;
2743 			init_waitqueue_head(&my_qpair->event);
2744 			wakeup_cb = qp_wakeup_cb;
2745 			client_data = (void *)my_qpair;
2746 		}
2747 	} else {
2748 		my_qpair->guest_endpoint = true;
2749 	}
2750 
2751 	retval = vmci_qp_alloc(handle,
2752 			       &my_qpair->produce_q,
2753 			       my_qpair->produce_q_size,
2754 			       &my_qpair->consume_q,
2755 			       my_qpair->consume_q_size,
2756 			       my_qpair->peer,
2757 			       my_qpair->flags,
2758 			       my_qpair->priv_flags,
2759 			       my_qpair->guest_endpoint,
2760 			       wakeup_cb, client_data);
2761 
2762 	if (retval < VMCI_SUCCESS) {
2763 		kfree(my_qpair);
2764 		return retval;
2765 	}
2766 
2767 	*qpair = my_qpair;
2768 	my_qpair->handle = *handle;
2769 
2770 	return retval;
2771 }
2772 EXPORT_SYMBOL_GPL(vmci_qpair_alloc);
2773 
2774 /*
2775  * vmci_qpair_detach() - Detatches the client from a queue pair.
2776  * @qpair:      Reference of a pointer to the qpair struct.
2777  *
2778  * This is the client interface for detaching from a VMCIQPair.
2779  * Note that this routine will free the memory allocated for the
2780  * vmci_qp structure too.
2781  */
2782 int vmci_qpair_detach(struct vmci_qp **qpair)
2783 {
2784 	int result;
2785 	struct vmci_qp *old_qpair;
2786 
2787 	if (!qpair || !(*qpair))
2788 		return VMCI_ERROR_INVALID_ARGS;
2789 
2790 	old_qpair = *qpair;
2791 	result = qp_detatch(old_qpair->handle, old_qpair->guest_endpoint);
2792 
2793 	/*
2794 	 * The guest can fail to detach for a number of reasons, and
2795 	 * if it does so, it will cleanup the entry (if there is one).
2796 	 * The host can fail too, but it won't cleanup the entry
2797 	 * immediately, it will do that later when the context is
2798 	 * freed.  Either way, we need to release the qpair struct
2799 	 * here; there isn't much the caller can do, and we don't want
2800 	 * to leak.
2801 	 */
2802 
2803 	memset(old_qpair, 0, sizeof(*old_qpair));
2804 	old_qpair->handle = VMCI_INVALID_HANDLE;
2805 	old_qpair->peer = VMCI_INVALID_ID;
2806 	kfree(old_qpair);
2807 	*qpair = NULL;
2808 
2809 	return result;
2810 }
2811 EXPORT_SYMBOL_GPL(vmci_qpair_detach);
2812 
2813 /*
2814  * vmci_qpair_get_produce_indexes() - Retrieves the indexes of the producer.
2815  * @qpair:      Pointer to the queue pair struct.
2816  * @producer_tail:      Reference used for storing producer tail index.
2817  * @consumer_head:      Reference used for storing the consumer head index.
2818  *
2819  * This is the client interface for getting the current indexes of the
2820  * QPair from the point of the view of the caller as the producer.
2821  */
2822 int vmci_qpair_get_produce_indexes(const struct vmci_qp *qpair,
2823 				   u64 *producer_tail,
2824 				   u64 *consumer_head)
2825 {
2826 	struct vmci_queue_header *produce_q_header;
2827 	struct vmci_queue_header *consume_q_header;
2828 	int result;
2829 
2830 	if (!qpair)
2831 		return VMCI_ERROR_INVALID_ARGS;
2832 
2833 	qp_lock(qpair);
2834 	result =
2835 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2836 	if (result == VMCI_SUCCESS)
2837 		vmci_q_header_get_pointers(produce_q_header, consume_q_header,
2838 					   producer_tail, consumer_head);
2839 	qp_unlock(qpair);
2840 
2841 	if (result == VMCI_SUCCESS &&
2842 	    ((producer_tail && *producer_tail >= qpair->produce_q_size) ||
2843 	     (consumer_head && *consumer_head >= qpair->produce_q_size)))
2844 		return VMCI_ERROR_INVALID_SIZE;
2845 
2846 	return result;
2847 }
2848 EXPORT_SYMBOL_GPL(vmci_qpair_get_produce_indexes);
2849 
2850 /*
2851  * vmci_qpair_get_consume_indexes() - Retrieves the indexes of the consumer.
2852  * @qpair:      Pointer to the queue pair struct.
2853  * @consumer_tail:      Reference used for storing consumer tail index.
2854  * @producer_head:      Reference used for storing the producer head index.
2855  *
2856  * This is the client interface for getting the current indexes of the
2857  * QPair from the point of the view of the caller as the consumer.
2858  */
2859 int vmci_qpair_get_consume_indexes(const struct vmci_qp *qpair,
2860 				   u64 *consumer_tail,
2861 				   u64 *producer_head)
2862 {
2863 	struct vmci_queue_header *produce_q_header;
2864 	struct vmci_queue_header *consume_q_header;
2865 	int result;
2866 
2867 	if (!qpair)
2868 		return VMCI_ERROR_INVALID_ARGS;
2869 
2870 	qp_lock(qpair);
2871 	result =
2872 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2873 	if (result == VMCI_SUCCESS)
2874 		vmci_q_header_get_pointers(consume_q_header, produce_q_header,
2875 					   consumer_tail, producer_head);
2876 	qp_unlock(qpair);
2877 
2878 	if (result == VMCI_SUCCESS &&
2879 	    ((consumer_tail && *consumer_tail >= qpair->consume_q_size) ||
2880 	     (producer_head && *producer_head >= qpair->consume_q_size)))
2881 		return VMCI_ERROR_INVALID_SIZE;
2882 
2883 	return result;
2884 }
2885 EXPORT_SYMBOL_GPL(vmci_qpair_get_consume_indexes);
2886 
2887 /*
2888  * vmci_qpair_produce_free_space() - Retrieves free space in producer queue.
2889  * @qpair:      Pointer to the queue pair struct.
2890  *
2891  * This is the client interface for getting the amount of free
2892  * space in the QPair from the point of the view of the caller as
2893  * the producer which is the common case.  Returns < 0 if err, else
2894  * available bytes into which data can be enqueued if > 0.
2895  */
2896 s64 vmci_qpair_produce_free_space(const struct vmci_qp *qpair)
2897 {
2898 	struct vmci_queue_header *produce_q_header;
2899 	struct vmci_queue_header *consume_q_header;
2900 	s64 result;
2901 
2902 	if (!qpair)
2903 		return VMCI_ERROR_INVALID_ARGS;
2904 
2905 	qp_lock(qpair);
2906 	result =
2907 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2908 	if (result == VMCI_SUCCESS)
2909 		result = vmci_q_header_free_space(produce_q_header,
2910 						  consume_q_header,
2911 						  qpair->produce_q_size);
2912 	else
2913 		result = 0;
2914 
2915 	qp_unlock(qpair);
2916 
2917 	return result;
2918 }
2919 EXPORT_SYMBOL_GPL(vmci_qpair_produce_free_space);
2920 
2921 /*
2922  * vmci_qpair_consume_free_space() - Retrieves free space in consumer queue.
2923  * @qpair:      Pointer to the queue pair struct.
2924  *
2925  * This is the client interface for getting the amount of free
2926  * space in the QPair from the point of the view of the caller as
2927  * the consumer which is not the common case.  Returns < 0 if err, else
2928  * available bytes into which data can be enqueued if > 0.
2929  */
2930 s64 vmci_qpair_consume_free_space(const struct vmci_qp *qpair)
2931 {
2932 	struct vmci_queue_header *produce_q_header;
2933 	struct vmci_queue_header *consume_q_header;
2934 	s64 result;
2935 
2936 	if (!qpair)
2937 		return VMCI_ERROR_INVALID_ARGS;
2938 
2939 	qp_lock(qpair);
2940 	result =
2941 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2942 	if (result == VMCI_SUCCESS)
2943 		result = vmci_q_header_free_space(consume_q_header,
2944 						  produce_q_header,
2945 						  qpair->consume_q_size);
2946 	else
2947 		result = 0;
2948 
2949 	qp_unlock(qpair);
2950 
2951 	return result;
2952 }
2953 EXPORT_SYMBOL_GPL(vmci_qpair_consume_free_space);
2954 
2955 /*
2956  * vmci_qpair_produce_buf_ready() - Gets bytes ready to read from
2957  * producer queue.
2958  * @qpair:      Pointer to the queue pair struct.
2959  *
2960  * This is the client interface for getting the amount of
2961  * enqueued data in the QPair from the point of the view of the
2962  * caller as the producer which is not the common case.  Returns < 0 if err,
2963  * else available bytes that may be read.
2964  */
2965 s64 vmci_qpair_produce_buf_ready(const struct vmci_qp *qpair)
2966 {
2967 	struct vmci_queue_header *produce_q_header;
2968 	struct vmci_queue_header *consume_q_header;
2969 	s64 result;
2970 
2971 	if (!qpair)
2972 		return VMCI_ERROR_INVALID_ARGS;
2973 
2974 	qp_lock(qpair);
2975 	result =
2976 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
2977 	if (result == VMCI_SUCCESS)
2978 		result = vmci_q_header_buf_ready(produce_q_header,
2979 						 consume_q_header,
2980 						 qpair->produce_q_size);
2981 	else
2982 		result = 0;
2983 
2984 	qp_unlock(qpair);
2985 
2986 	return result;
2987 }
2988 EXPORT_SYMBOL_GPL(vmci_qpair_produce_buf_ready);
2989 
2990 /*
2991  * vmci_qpair_consume_buf_ready() - Gets bytes ready to read from
2992  * consumer queue.
2993  * @qpair:      Pointer to the queue pair struct.
2994  *
2995  * This is the client interface for getting the amount of
2996  * enqueued data in the QPair from the point of the view of the
2997  * caller as the consumer which is the normal case.  Returns < 0 if err,
2998  * else available bytes that may be read.
2999  */
3000 s64 vmci_qpair_consume_buf_ready(const struct vmci_qp *qpair)
3001 {
3002 	struct vmci_queue_header *produce_q_header;
3003 	struct vmci_queue_header *consume_q_header;
3004 	s64 result;
3005 
3006 	if (!qpair)
3007 		return VMCI_ERROR_INVALID_ARGS;
3008 
3009 	qp_lock(qpair);
3010 	result =
3011 	    qp_get_queue_headers(qpair, &produce_q_header, &consume_q_header);
3012 	if (result == VMCI_SUCCESS)
3013 		result = vmci_q_header_buf_ready(consume_q_header,
3014 						 produce_q_header,
3015 						 qpair->consume_q_size);
3016 	else
3017 		result = 0;
3018 
3019 	qp_unlock(qpair);
3020 
3021 	return result;
3022 }
3023 EXPORT_SYMBOL_GPL(vmci_qpair_consume_buf_ready);
3024 
3025 /*
3026  * vmci_qpair_enqueue() - Throw data on the queue.
3027  * @qpair:      Pointer to the queue pair struct.
3028  * @buf:        Pointer to buffer containing data
3029  * @buf_size:   Length of buffer.
3030  * @buf_type:   Buffer type (Unused).
3031  *
3032  * This is the client interface for enqueueing data into the queue.
3033  * Returns number of bytes enqueued or < 0 on error.
3034  */
3035 ssize_t vmci_qpair_enqueue(struct vmci_qp *qpair,
3036 			   const void *buf,
3037 			   size_t buf_size,
3038 			   int buf_type)
3039 {
3040 	ssize_t result;
3041 	struct iov_iter from;
3042 	struct kvec v = {.iov_base = (void *)buf, .iov_len = buf_size};
3043 
3044 	if (!qpair || !buf)
3045 		return VMCI_ERROR_INVALID_ARGS;
3046 
3047 	iov_iter_kvec(&from, ITER_SOURCE, &v, 1, buf_size);
3048 
3049 	qp_lock(qpair);
3050 
3051 	do {
3052 		result = qp_enqueue_locked(qpair->produce_q,
3053 					   qpair->consume_q,
3054 					   qpair->produce_q_size,
3055 					   &from);
3056 
3057 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3058 		    !qp_wait_for_ready_queue(qpair))
3059 			result = VMCI_ERROR_WOULD_BLOCK;
3060 
3061 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3062 
3063 	qp_unlock(qpair);
3064 
3065 	return result;
3066 }
3067 EXPORT_SYMBOL_GPL(vmci_qpair_enqueue);
3068 
3069 /*
3070  * vmci_qpair_dequeue() - Get data from the queue.
3071  * @qpair:      Pointer to the queue pair struct.
3072  * @buf:        Pointer to buffer for the data
3073  * @buf_size:   Length of buffer.
3074  * @buf_type:   Buffer type (Unused).
3075  *
3076  * This is the client interface for dequeueing data from the queue.
3077  * Returns number of bytes dequeued or < 0 on error.
3078  */
3079 ssize_t vmci_qpair_dequeue(struct vmci_qp *qpair,
3080 			   void *buf,
3081 			   size_t buf_size,
3082 			   int buf_type)
3083 {
3084 	ssize_t result;
3085 	struct iov_iter to;
3086 	struct kvec v = {.iov_base = buf, .iov_len = buf_size};
3087 
3088 	if (!qpair || !buf)
3089 		return VMCI_ERROR_INVALID_ARGS;
3090 
3091 	iov_iter_kvec(&to, ITER_DEST, &v, 1, buf_size);
3092 
3093 	qp_lock(qpair);
3094 
3095 	do {
3096 		result = qp_dequeue_locked(qpair->produce_q,
3097 					   qpair->consume_q,
3098 					   qpair->consume_q_size,
3099 					   &to, true);
3100 
3101 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3102 		    !qp_wait_for_ready_queue(qpair))
3103 			result = VMCI_ERROR_WOULD_BLOCK;
3104 
3105 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3106 
3107 	qp_unlock(qpair);
3108 
3109 	return result;
3110 }
3111 EXPORT_SYMBOL_GPL(vmci_qpair_dequeue);
3112 
3113 /*
3114  * vmci_qpair_peek() - Peek at the data in the queue.
3115  * @qpair:      Pointer to the queue pair struct.
3116  * @buf:        Pointer to buffer for the data
3117  * @buf_size:   Length of buffer.
3118  * @buf_type:   Buffer type (Unused on Linux).
3119  *
3120  * This is the client interface for peeking into a queue.  (I.e.,
3121  * copy data from the queue without updating the head pointer.)
3122  * Returns number of bytes dequeued or < 0 on error.
3123  */
3124 ssize_t vmci_qpair_peek(struct vmci_qp *qpair,
3125 			void *buf,
3126 			size_t buf_size,
3127 			int buf_type)
3128 {
3129 	struct iov_iter to;
3130 	struct kvec v = {.iov_base = buf, .iov_len = buf_size};
3131 	ssize_t result;
3132 
3133 	if (!qpair || !buf)
3134 		return VMCI_ERROR_INVALID_ARGS;
3135 
3136 	iov_iter_kvec(&to, ITER_DEST, &v, 1, buf_size);
3137 
3138 	qp_lock(qpair);
3139 
3140 	do {
3141 		result = qp_dequeue_locked(qpair->produce_q,
3142 					   qpair->consume_q,
3143 					   qpair->consume_q_size,
3144 					   &to, false);
3145 
3146 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3147 		    !qp_wait_for_ready_queue(qpair))
3148 			result = VMCI_ERROR_WOULD_BLOCK;
3149 
3150 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3151 
3152 	qp_unlock(qpair);
3153 
3154 	return result;
3155 }
3156 EXPORT_SYMBOL_GPL(vmci_qpair_peek);
3157 
3158 /*
3159  * vmci_qpair_enquev() - Throw data on the queue using iov.
3160  * @qpair:      Pointer to the queue pair struct.
3161  * @iov:        Pointer to buffer containing data
3162  * @iov_size:   Length of buffer.
3163  * @buf_type:   Buffer type (Unused).
3164  *
3165  * This is the client interface for enqueueing data into the queue.
3166  * This function uses IO vectors to handle the work. Returns number
3167  * of bytes enqueued or < 0 on error.
3168  */
3169 ssize_t vmci_qpair_enquev(struct vmci_qp *qpair,
3170 			  struct msghdr *msg,
3171 			  size_t iov_size,
3172 			  int buf_type)
3173 {
3174 	ssize_t result;
3175 
3176 	if (!qpair)
3177 		return VMCI_ERROR_INVALID_ARGS;
3178 
3179 	qp_lock(qpair);
3180 
3181 	do {
3182 		result = qp_enqueue_locked(qpair->produce_q,
3183 					   qpair->consume_q,
3184 					   qpair->produce_q_size,
3185 					   &msg->msg_iter);
3186 
3187 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3188 		    !qp_wait_for_ready_queue(qpair))
3189 			result = VMCI_ERROR_WOULD_BLOCK;
3190 
3191 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3192 
3193 	qp_unlock(qpair);
3194 
3195 	return result;
3196 }
3197 EXPORT_SYMBOL_GPL(vmci_qpair_enquev);
3198 
3199 /*
3200  * vmci_qpair_dequev() - Get data from the queue using iov.
3201  * @qpair:      Pointer to the queue pair struct.
3202  * @iov:        Pointer to buffer for the data
3203  * @iov_size:   Length of buffer.
3204  * @buf_type:   Buffer type (Unused).
3205  *
3206  * This is the client interface for dequeueing data from the queue.
3207  * This function uses IO vectors to handle the work. Returns number
3208  * of bytes dequeued or < 0 on error.
3209  */
3210 ssize_t vmci_qpair_dequev(struct vmci_qp *qpair,
3211 			  struct msghdr *msg,
3212 			  size_t iov_size,
3213 			  int buf_type)
3214 {
3215 	ssize_t result;
3216 
3217 	if (!qpair)
3218 		return VMCI_ERROR_INVALID_ARGS;
3219 
3220 	qp_lock(qpair);
3221 
3222 	do {
3223 		result = qp_dequeue_locked(qpair->produce_q,
3224 					   qpair->consume_q,
3225 					   qpair->consume_q_size,
3226 					   &msg->msg_iter, true);
3227 
3228 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3229 		    !qp_wait_for_ready_queue(qpair))
3230 			result = VMCI_ERROR_WOULD_BLOCK;
3231 
3232 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3233 
3234 	qp_unlock(qpair);
3235 
3236 	return result;
3237 }
3238 EXPORT_SYMBOL_GPL(vmci_qpair_dequev);
3239 
3240 /*
3241  * vmci_qpair_peekv() - Peek at the data in the queue using iov.
3242  * @qpair:      Pointer to the queue pair struct.
3243  * @iov:        Pointer to buffer for the data
3244  * @iov_size:   Length of buffer.
3245  * @buf_type:   Buffer type (Unused on Linux).
3246  *
3247  * This is the client interface for peeking into a queue.  (I.e.,
3248  * copy data from the queue without updating the head pointer.)
3249  * This function uses IO vectors to handle the work. Returns number
3250  * of bytes peeked or < 0 on error.
3251  */
3252 ssize_t vmci_qpair_peekv(struct vmci_qp *qpair,
3253 			 struct msghdr *msg,
3254 			 size_t iov_size,
3255 			 int buf_type)
3256 {
3257 	ssize_t result;
3258 
3259 	if (!qpair)
3260 		return VMCI_ERROR_INVALID_ARGS;
3261 
3262 	qp_lock(qpair);
3263 
3264 	do {
3265 		result = qp_dequeue_locked(qpair->produce_q,
3266 					   qpair->consume_q,
3267 					   qpair->consume_q_size,
3268 					   &msg->msg_iter, false);
3269 
3270 		if (result == VMCI_ERROR_QUEUEPAIR_NOT_READY &&
3271 		    !qp_wait_for_ready_queue(qpair))
3272 			result = VMCI_ERROR_WOULD_BLOCK;
3273 
3274 	} while (result == VMCI_ERROR_QUEUEPAIR_NOT_READY);
3275 
3276 	qp_unlock(qpair);
3277 	return result;
3278 }
3279 EXPORT_SYMBOL_GPL(vmci_qpair_peekv);
3280