Implement queueInode and queueFD in mqfs.

Implement inode and file description representing a POSIX message queue,
and other utilities needed to implement file operations.

Updates #136

pkg/sentry/fsimpl/kernfs/dynamic_bytes_file.go

@@ -74,6 +74,11 @@ func (*DynamicBytesFile) SetStat(context.Context, *vfs.Filesystem, *auth.Credent
 	return linuxerr.EPERM
 }
 
+// Locks returns the file locks for this file.
+func (f *DynamicBytesFile) Locks() *vfs.FileLocks {
+	return &f.locks
+}
+
 // DynamicBytesFD implements vfs.FileDescriptionImpl for an FD backed by a
 // DynamicBytesFile.
 //

pkg/sentry/fsimpl/mqfs/BUILD

@@ -18,7 +18,8 @@ go_library(
     name = "mqfs",
     srcs = [
         "mqfs.go",
-        "inodes.go",
+        "root.go",
+        "queue.go",
         "root_inode_refs.go",
     ],
     visibility = ["//pkg/sentry:internal"],
@@ -29,6 +30,9 @@ go_library(
         "//pkg/refsvfs2",
         "//pkg/sentry/fsimpl/kernfs",
         "//pkg/sentry/kernel/auth",
+        "//pkg/sentry/kernel/mq",
         "//pkg/sentry/vfs",
+        "//pkg/usermem",
+        "//pkg/waiter",
     ],
 )

pkg/sentry/fsimpl/mqfs/queue.go

@@ -0,0 +1,145 @@
+// Copyright 2021 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mqfs
+
+import (
+	"gvisor.dev/gvisor/pkg/abi/linux"
+	"gvisor.dev/gvisor/pkg/context"
+	"gvisor.dev/gvisor/pkg/errors/linuxerr"
+	"gvisor.dev/gvisor/pkg/sentry/fsimpl/kernfs"
+	"gvisor.dev/gvisor/pkg/sentry/kernel/auth"
+	"gvisor.dev/gvisor/pkg/sentry/kernel/mq"
+	"gvisor.dev/gvisor/pkg/sentry/vfs"
+	"gvisor.dev/gvisor/pkg/usermem"
+	"gvisor.dev/gvisor/pkg/waiter"
+)
+
+// queueInode represents an inode for a message queue (/dev/mqueue/[name]).
+//
+// +stateify savable
+type queueInode struct {
+	kernfs.DynamicBytesFile
+
+	// queue is the message queue backing this inode.
+	queue *mq.Queue
+}
+
+var _ kernfs.Inode = (*queueInode)(nil)
+
+// newQueueInode returns a new, initialized queueInode.
+func (fs *filesystem) newQueueInode(ctx context.Context, creds *auth.Credentials, q *mq.Queue, perm linux.FileMode) kernfs.Inode {
+	inode := &queueInode{queue: q}
+	inode.Init(ctx, creds, linux.UNNAMED_MAJOR, fs.devMinor, fs.NextIno(), q, perm)
+	return inode
+}
+
+// Keep implements kernfs.Inode.Keep.
+func (q *queueInode) Keep() bool {
+	// Return true so that the fs keeps newly created dentries. This is done
+	// because inodes returned by root.Lookup are not temporary, they exist
+	// in the fs, and refer to message queues.
+	return true
+}
+
+// queueFD implements vfs.FileDescriptionImpl for FD backed by a POSIX message
+// queue. It's mostly similar to DynamicBytesFD, but implements more operations.
+//
+// +stateify savable
+type queueFD struct {
+	vfs.FileDescriptionDefaultImpl
+	vfs.DynamicBytesFileDescriptionImpl
+	vfs.LockFD
+
+	vfsfd vfs.FileDescription
+	inode kernfs.Inode
+
+	// queue is the queue backing this fd.
+	queue *mq.Queue
+}
+
+// Init initializes a queueFD. Mostly copied from DynamicBytesFD.Init, but uses
+// the queueFD as FileDescriptionImpl.
+func (fd *queueFD) Init(m *vfs.Mount, d *kernfs.Dentry, data vfs.DynamicBytesSource, locks *vfs.FileLocks, flags uint32) error {
+	fd.LockFD.Init(locks)
+	if err := fd.vfsfd.Init(fd, flags, m, d.VFSDentry(), &vfs.FileDescriptionOptions{}); err != nil {
+		return err
+	}
+	fd.inode = d.Inode()
+	fd.SetDataSource(data)
+	return nil
+}
+
+// Seek implements vfs.FileDescriptionImpl.Seek.
+func (fd *queueFD) Seek(ctx context.Context, offset int64, whence int32) (int64, error) {
+	return fd.DynamicBytesFileDescriptionImpl.Seek(ctx, offset, whence)
+}
+
+// Read implements vfs.FileDescriptionImpl.Read.
+func (fd *queueFD) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.ReadOptions) (int64, error) {
+	return fd.DynamicBytesFileDescriptionImpl.Read(ctx, dst, opts)
+}
+
+// PRead implements vfs.FileDescriptionImpl.PRead.
+func (fd *queueFD) PRead(ctx context.Context, dst usermem.IOSequence, offset int64, opts vfs.ReadOptions) (int64, error) {
+	return fd.DynamicBytesFileDescriptionImpl.PRead(ctx, dst, offset, opts)
+}
+
+// Write implements vfs.FileDescriptionImpl.Write.
+func (fd *queueFD) Write(ctx context.Context, src usermem.IOSequence, opts vfs.WriteOptions) (int64, error) {
+	return fd.DynamicBytesFileDescriptionImpl.Write(ctx, src, opts)
+}
+
+// PWrite implements vfs.FileDescriptionImpl.PWrite.
+func (fd *queueFD) PWrite(ctx context.Context, src usermem.IOSequence, offset int64, opts vfs.WriteOptions) (int64, error) {
+	return fd.DynamicBytesFileDescriptionImpl.PWrite(ctx, src, offset, opts)
+}
+
+// Release implements vfs.FileDescriptionImpl.Release.
+func (fd *queueFD) Release(context.Context) {}
+
+// Stat implements vfs.FileDescriptionImpl.Stat.
+func (fd *queueFD) Stat(ctx context.Context, opts vfs.StatOptions) (linux.Statx, error) {
+	fs := fd.vfsfd.VirtualDentry().Mount().Filesystem()
+	return fd.inode.Stat(ctx, fs, opts)
+}
+
+// SetStat implements vfs.FileDescriptionImpl.SetStat.
+func (fd *queueFD) SetStat(context.Context, vfs.SetStatOptions) error {
+	// DynamicBytesFiles are immutable.
+	return linuxerr.EPERM
+}
+
+// OnClose implements FileDescriptionImpl.OnClose similar to
+// ipc/mqueue.c::mqueue_flush_file.
+func (fd *queueFD) OnClose(ctx context.Context) error {
+	fd.queue.Flush(ctx)
+	return nil
+}
+
+// Readiness implements waiter.Waitable.Readiness similar to
+// ipc/mqueue.c::mqueue_poll_file.
+func (fd *queueFD) Readiness(mask waiter.EventMask) waiter.EventMask {
+	return fd.queue.Readiness(mask)
+}
+
+// EventRegister implements Waitable.EventRegister.
+func (fd *queueFD) EventRegister(e *waiter.Entry, mask waiter.EventMask) {
+	fd.queue.EventRegister(e, mask)
+}
+
+// EventUnregister implements Waitable.EventUnregister.
+func (fd *queueFD) EventUnregister(e *waiter.Entry) {
+	fd.queue.EventUnregister(e)
+}

pkg/sentry/fsimpl/mqfs/root.go

@@ -15,8 +15,6 @@
 package mqfs
 
 import (
-	"bytes"
-
 	"gvisor.dev/gvisor/pkg/abi/linux"
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"

pkg/sentry/kernel/mq/BUILD

@@ -24,6 +24,7 @@ go_library(
     visibility = ["//pkg/sentry:internal"],
     deps = [
         "//pkg/abi/linux",
+        "//pkg/context",
         "//pkg/sentry/fs",
         "//pkg/sync",
         "//pkg/waiter",

pkg/sentry/kernel/mq/mq.go

@@ -16,7 +16,11 @@
 package mq
 
 import (
+	"bytes"
+	"fmt"
+
 	"gvisor.dev/gvisor/pkg/abi/linux"
+	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/sentry/fs"
 	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/waiter"
@@ -52,7 +56,7 @@ type Queue struct {
 
 	// subscriber represents a task registered to receive async notification
 	// from this queue.
-	subscriber Subscriber
+	subscriber *Subscriber
 
 	// nonBlock is true if this queue is non-blocking.
 	nonBlock bool
@@ -93,4 +97,82 @@ type Message struct {
 // +stateify savable
 type Subscriber struct {
 	// TODO: Add fields when mq_notify(2) is implemented.
+
+	// pid is the PID of the registered task.
+	pid int32
+}
+
+// Generate implements vfs.DynamicBytesSource.Generate. Queue is used as a
+// dynamic bytes source for mqfs's queueInode.
+func (q *Queue) Generate(ctx context.Context, buf *bytes.Buffer) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	var (
+		pid       int32
+		method    int
+		sigNumber int
+	)
+	if q.subscriber != nil {
+		pid = q.subscriber.pid
+		// TODO: add method and sigNumber when mq_notify(2) is implemented.
+	}
+
+	buf.WriteString(
+		fmt.Sprintf("QSIZE:%-10d NOTIFY:%-5d SIGNO:%-5d NOTIFY_PID:%-6d\n",
+			q.byteCount, method, sigNumber, pid),
+	)
+	return nil
+}
+
+// Flush checks if the calling process has attached a notification request to
+// this queue, if yes, then the request is removed, and another process can
+// attach a request.
+func (q *Queue) Flush(ctx context.Context) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	pid, ok := context.ThreadGroupIDFromContext(ctx)
+	if ok {
+		if q.subscriber != nil && pid == q.subscriber.pid {
+			q.subscriber = nil
+		}
+	}
+}
+
+// Readiness implements Waitable.Readiness.
+func (q *Queue) Readiness(mask waiter.EventMask) waiter.EventMask {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	events := waiter.EventMask(0)
+	if q.messageCount > 0 {
+		events |= waiter.ReadableEvents
+	}
+	if q.messageCount < q.maxMessageCount {
+		events |= waiter.WritableEvents
+	}
+	return events & mask
+}
+
+// EventRegister implements Waitable.EventRegister.
+func (q *Queue) EventRegister(e *waiter.Entry, mask waiter.EventMask) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	if mask&waiter.WritableEvents != 0 {
+		q.senders.EventRegister(e, waiter.EventOut)
+	}
+	if mask&waiter.ReadableEvents != 0 {
+		q.receivers.EventRegister(e, waiter.EventIn)
+	}
+}
+
+// EventUnregister implements Waitable.EventUnregister.
+func (q *Queue) EventUnregister(e *waiter.Entry) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.senders.EventUnregister(e)
+	q.receivers.EventUnregister(e)
 }