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(dev: &vmci_pdev->dev, PAGE_SIZE,
258	cpu_addr: queue->kernel_if->u.g.vas[i],
259	dma_handle: queue->kernel_if->u.g.pas[i]);
260	}
261
262	vfree(addr: 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(size: 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(dev: &vmci_pdev->dev, PAGE_SIZE,
310	dma_handle: &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(q: queue, size: 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(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(addr: (u8 *)va + page_offset, bytes: to_copy,
360	i: 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(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(addr: (u8 *)va + page_offset, bytes: to_copy, i: 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(n: num_produce_pages, size: sizeof(*produce_ppns),
447	GFP_KERNEL);
448	if (!produce_ppns)
449	return VMCI_ERROR_NO_MEM;
450
451	consume_ppns =
452	kmalloc_array(n: num_consume_pages, size: sizeof(*consume_ppns),
453	GFP_KERNEL);
454	if (!consume_ppns) {
455	kfree(objp: 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(objp: ppn_set->produce_ppns);
483	kfree(objp: 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(size: 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(objp: 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(lock: 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(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(start: (uintptr_t) produce_uva,
659	nr_pages: produce_q->kernel_if->num_pages,
660	gup_flags: FOLL_WRITE,
661	pages: 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(pages: produce_q->kernel_if->u.h.header_page,
667	num_pages: retval, dirty: false);
668	err = VMCI_ERROR_NO_MEM;
669	goto out;
670	}
671
672	retval = get_user_pages_fast(start: (uintptr_t) consume_uva,
673	nr_pages: consume_q->kernel_if->num_pages,
674	gup_flags: FOLL_WRITE,
675	pages: 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(pages: consume_q->kernel_if->u.h.header_page,
681	num_pages: retval, dirty: false);
682	qp_release_pages(pages: produce_q->kernel_if->u.h.header_page,
683	num_pages: produce_q->kernel_if->num_pages, dirty: 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(pages: produce_q->kernel_if->u.h.header_page,
724	num_pages: produce_q->kernel_if->num_pages, dirty: 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(pages: consume_q->kernel_if->u.h.header_page,
729	num_pages: consume_q->kernel_if->num_pages, dirty: 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(pages: headers, count: 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(addr: produce_q->q_header);
790	else
791	vunmap(addr: 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(h: handle))
810	return NULL;
811
812	list_for_each_entry(entry, &qp_list->head, list_item) {
813	if (vmci_handle_is_equal(h1: entry->handle, h2: 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_list: &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_list: &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(msg: &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(h: 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(size: 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(list: &entry->qp.list_item);
909
910	/* Add resource obj */
911	result = vmci_resource_add(resource: &entry->resource,
912	resource_type: VMCI_RESOURCE_TYPE_QPAIR_GUEST,
913	handle);
914	entry->qp.handle = vmci_resource_handle(resource: &entry->resource);
915	if ((result != VMCI_SUCCESS) ||
916	qp_list_find(qp_list: &qp_guest_endpoints, handle: 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(objp: 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(ppn_set: &entry->ppn_set);
932	qp_cleanup_queue_mutex(produce_q: entry->produce_q, consume_q: entry->consume_q);
933	qp_free_queue(q: entry->produce_q, size: entry->qp.produce_size);
934	qp_free_queue(q: entry->consume_q, size: entry->qp.consume_size);
935	/* Unlink from resource hash table and free callback */
936	vmci_resource_remove(resource: &entry->resource);
937
938	kfree(objp: 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(size: 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(call_buf: (u8 *)alloc_msg + sizeof(*alloc_msg),
974	ppn_set: &entry->ppn_set);
975	if (result == VMCI_SUCCESS)
976	result = vmci_send_datagram(dg: &alloc_msg->hdr);
977
978	kfree(objp: 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(dg: &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(new: &entry->list_item, head: &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: &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(lock: &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(attach: 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(lock: &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_list: &qp_guest_endpoints, entry: &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(lock: &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: *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(attach: true, handle: *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(size: 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(size: 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: *handle, peer, flags,
1169	produce_size, consume_size,
1170	produce_q: my_produce_q, consume_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(prod_q: my_produce_q, num_produce_pages, cons_q: my_consume_q,
1178	num_consume_pages,
1179	ppn_set: &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(entry: 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(produce_q: (struct vmci_queue *)my_produce_q,
1222	consume_q: (struct vmci_queue *)my_consume_q);
1223
1224	qp_list_add_entry(qp_list: &qp_guest_endpoints, entry: &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(q_header: (*produce_q)->q_header, handle: *handle);
1240	vmci_q_header_init(q_header: (*consume_q)->q_header, handle: *handle);
1241	}
1242
1243	mutex_unlock(lock: &qp_guest_endpoints.mutex);
1244
1245	return VMCI_SUCCESS;
1246
1247	error:
1248	mutex_unlock(lock: &qp_guest_endpoints.mutex);
1249	if (queue_pair_entry) {
1250	/* The queues will be freed inside the destroy routine. */
1251	qp_guest_endpoint_destroy(entry: queue_pair_entry);
1252	} else {
1253	qp_free_queue(q: my_produce_q, size: produce_size);
1254	qp_free_queue(q: my_consume_q, size: 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(lock: &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(size: 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(size: 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(size: 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(produce_q: entry->produce_q, consume_q: entry->consume_q);
1369
1370	INIT_LIST_HEAD(list: &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	produce_q: entry->produce_q,
1393	consume_q: 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_list: &qp_broker_list, entry: &entry->qp);
1410	if (ent != NULL)
1411	*ent = entry;
1412
1413	/* Add to resource obj */
1414	result = vmci_resource_add(resource: &entry->resource,
1415	resource_type: 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(resource: &entry->resource);
1424	if (is_local) {
1425	vmci_q_header_init(q_header: entry->produce_q->q_header,
1426	handle: entry->qp.handle);
1427	vmci_q_header_init(q_header: entry->consume_q->q_header,
1428	handle: entry->qp.handle);
1429	}
1430
1431	vmci_ctx_qp_create(context, handle: entry->qp.handle);
1432
1433	return VMCI_SUCCESS;
1434
1435	error:
1436	if (entry != NULL) {
1437	qp_host_free_queue(queue: entry->produce_q, queue_size: guest_produce_size);
1438	qp_host_free_queue(queue: entry->consume_q, queue_size: guest_consume_size);
1439	kfree(objp: 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(h: 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	dg: &ev.msg.hdr, from_guest: 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(cid: entry->create_id);
1587	supports_host_qp = vmci_ctx_supports_host_qp(context: create_context);
1588	vmci_ctx_put(context: 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	produce_q: entry->produce_q,
1640	consume_q: 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(attach: true, handle: entry->qp.handle, my_id: context_id,
1665	peer_id: 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, handle: 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(h: 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(lock: &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(lock: &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(h: *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(handle: new_handle, peer, flags, priv_flags,
1797	produce_size, consume_size, NULL, context,
1798	wakeup_cb, client_data, ent: &entry, swap: &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(resource: &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(h: 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(head: &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_list: &qp_broker_list))) {
1914	be = (struct qp_broker_entry *)entry;
1915
1916	qp_list_remove_entry(qp_list: &qp_broker_list, entry);
1917	kfree(objp: be);
1918	}
1919
1920	mutex_unlock(lock: &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(h: 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	produce_q: entry->produce_q, consume_q: entry->consume_q);
2020	if (result < VMCI_SUCCESS)
2021	goto out;
2022
2023	result = qp_host_map_queues(produce_q: entry->produce_q, consume_q: entry->consume_q);
2024	if (result < VMCI_SUCCESS) {
2025	qp_host_unregister_user_memory(produce_q: entry->produce_q,
2026	consume_q: 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(attach: true, handle, my_id: context_id, peer_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(lock: &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(h: 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(queue: 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	produce_q: entry->produce_q,
2146	consume_q: 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(produce_q: entry->produce_q,
2153	consume_q: entry->consume_q);
2154
2155	}
2156
2157	if (!headers_mapped)
2158	qp_reset_saved_headers(entry);
2159
2160	qp_release_queue_mutex(queue: 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_list: &qp_broker_list, entry: &entry->qp);
2174
2175	if (is_local)
2176	kfree(objp: entry->local_mem);
2177
2178	qp_cleanup_queue_mutex(produce_q: entry->produce_q, consume_q: entry->consume_q);
2179	qp_host_free_queue(queue: entry->produce_q, queue_size: entry->qp.produce_size);
2180	qp_host_free_queue(queue: entry->consume_q, queue_size: entry->qp.consume_size);
2181	/* Unlink from resource hash table and free callback */
2182	vmci_resource_remove(resource: &entry->resource);
2183
2184	kfree(objp: entry);
2185
2186	vmci_ctx_qp_destroy(context, handle);
2187	} else {
2188	qp_notify_peer(attach: false, handle, my_id: 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(lock: &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(h: 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(queue: entry->produce_q);
2256	qp_reset_saved_headers(entry);
2257	result =
2258	qp_host_register_user_memory(page_store: &page_store,
2259	produce_q: entry->produce_q,
2260	consume_q: entry->consume_q);
2261	qp_release_queue_mutex(queue: 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(lock: &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(produce_q: entry->produce_q, consume_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(h: 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(queue: 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, produce_q: entry->produce_q, consume_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(produce_q: entry->produce_q,
2374	consume_q: entry->consume_q);
2375
2376	/*
2377	* Move state from *_MEM to *_NO_MEM.
2378	*/
2379	entry->state--;
2380
2381	qp_release_queue_mutex(queue: entry->produce_q);
2382	}
2383
2384	result = VMCI_SUCCESS;
2385
2386	out:
2387	mutex_unlock(lock: &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_list: &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(handle: 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_list: &qp_guest_endpoints, entry);
2414
2415	qp_guest_endpoint_destroy(entry: ep);
2416	}
2417
2418	mutex_unlock(lock: &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(queue: 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(queue: 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(produce_q: qpair->produce_q, consume_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(i: 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_header: produce_q->q_header,
2556	consume_q_header: 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(q_header: produce_q->q_header);
2566	if (likely(tail + written < produce_q_size)) {
2567	result = qp_memcpy_to_queue_iter(queue: produce_q, queue_offset: tail, from, size: 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(queue: produce_q, queue_offset: tail, from, size: tmp);
2574	if (result >= VMCI_SUCCESS)
2575	result = qp_memcpy_to_queue_iter(queue: produce_q, queue_offset: 0, from,
2576	size: 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(q_header: produce_q->q_header, add: written,
2589	queue_size: 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(i: 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_header: consume_q->q_header,
2623	produce_q_header: 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(q_header: produce_q->q_header);
2639	if (likely(head + read < consume_q_size)) {
2640	result = qp_memcpy_from_queue_iter(to, queue: consume_q, queue_offset: head, size: 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, queue: consume_q, queue_offset: head, size: tmp);
2647	if (result >= VMCI_SUCCESS)
2648	result = qp_memcpy_from_queue_iter(to, queue: consume_q, queue_offset: 0,
2649	size: 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(q_header: produce_q->q_header,
2658	add: read, queue_size: 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: &src, dst: &dst, from_guest: false, route: &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(size: 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	produce_q: &my_qpair->produce_q,
2753	produce_size: my_qpair->produce_q_size,
2754	consume_q: &my_qpair->consume_q,
2755	consume_size: my_qpair->consume_q_size,
2756	peer: my_qpair->peer,
2757	flags: my_qpair->flags,
2758	priv_flags: my_qpair->priv_flags,
2759	guest_endpoint: my_qpair->guest_endpoint,
2760	wakeup_cb, client_data);
2761
2762	if (retval < VMCI_SUCCESS) {
2763	kfree(objp: 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(handle: old_qpair->handle, guest_endpoint: 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(objp: 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: &produce_q_header, consume_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: &produce_q_header, consume_q_header: &consume_q_header);
2873	if (result == VMCI_SUCCESS)
2874	vmci_q_header_get_pointers(produce_q_header: consume_q_header, consume_q_header: produce_q_header,
2875	producer_tail: consumer_tail, consumer_head: 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: &produce_q_header, consume_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	produce_q_size: 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: &produce_q_header, consume_q_header: &consume_q_header);
2942	if (result == VMCI_SUCCESS)
2943	result = vmci_q_header_free_space(produce_q_header: consume_q_header,
2944	consume_q_header: produce_q_header,
2945	produce_q_size: 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: &produce_q_header, consume_q_header: &consume_q_header);
2977	if (result == VMCI_SUCCESS)
2978	result = vmci_q_header_buf_ready(consume_q_header: produce_q_header,
2979	produce_q_header: consume_q_header,
2980	consume_q_size: 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: &produce_q_header, consume_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	consume_q_size: 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(i: &from, ITER_SOURCE, kvec: &v, nr_segs: 1, count: buf_size);
3048
3049	qp_lock(qpair);
3050
3051	do {
3052	result = qp_enqueue_locked(produce_q: qpair->produce_q,
3053	consume_q: qpair->consume_q,
3054	produce_q_size: qpair->produce_q_size,
3055	from: &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(i: &to, ITER_DEST, kvec: &v, nr_segs: 1, count: buf_size);
3092
3093	qp_lock(qpair);
3094
3095	do {
3096	result = qp_dequeue_locked(produce_q: qpair->produce_q,
3097	consume_q: qpair->consume_q,
3098	consume_q_size: qpair->consume_q_size,
3099	to: &to, update_consumer: 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(i: &to, ITER_DEST, kvec: &v, nr_segs: 1, count: buf_size);
3137
3138	qp_lock(qpair);
3139
3140	do {
3141	result = qp_dequeue_locked(produce_q: qpair->produce_q,
3142	consume_q: qpair->consume_q,
3143	consume_q_size: qpair->consume_q_size,
3144	to: &to, update_consumer: 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(produce_q: qpair->produce_q,
3183	consume_q: qpair->consume_q,
3184	produce_q_size: qpair->produce_q_size,
3185	from: &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(produce_q: qpair->produce_q,
3224	consume_q: qpair->consume_q,
3225	consume_q_size: qpair->consume_q_size,
3226	to: &msg->msg_iter, update_consumer: 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(produce_q: qpair->produce_q,
3266	consume_q: qpair->consume_q,
3267	consume_q_size: qpair->consume_q_size,
3268	to: &msg->msg_iter, update_consumer: 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