Python USES asyncio to encapsulate file reads and writes

preface

Like IO 1, reading and writing files is also a cumbersome operation.

By default, Python USES blocked reads and writes for the system. This means that in asyncio if called

f = file('xx')
f.read()

It blocks the event loop.

This article describes how to use the asyncio.Future object to encapsulate asynchronous reads and writes to a file.

Code in GitHub. Currently only Linux is supported.

Blocking and non-blocking

First, you need to change the read-write of the file to a non-blocking form. In the non-blocking case, each call to read returns immediately, if the return value is empty, meaning that the file operation has not been completed, or if the file contents have been read.

The switching between blocking and non-blocking is related to the operating system, so for the moment this article has only written the Linux version. If you have programming experience with the Unix system, you will find that the operation of Python is similar.

flag = fcntl.fcntl(self.fd, fcntl.F_GETFL) 
if fcntl.fcntl(self.fd, fcntl.F_SETFL, flag | os.O_NONBLOCK) != 0: 
  raise OSError() 

Future object

The Future object is similar to the Promise object in Javascript. It is a placeholder whose value will be calculated in the future. We can use

result = await future

Returns after future gets the value. While the use of

future.set_result(xxx)

You can set the value of future, which means that future can be returned. The await operator automatically calls future.result () to get the value.

loop.call_soon

You can insert a function into the event loop through the loop.call_soon method.

At this point, our asynchronous file read and write ideas are also out. Call the non-blocking read-write file function through loop.call_soon. If one file read and write is not completed, count the remaining bytes and insert the event loop again until the read and write is completed.

You can see that instead of the traditional while loop for non-blocking file reads and writes in Unix programming, the asyncio event loop is used.

The following is the schematic code for this 1 procedure.

def read_step(self, future, n, total):
  res = self.fd.read(n)
  if res is None:
    self.loop.call_soon(self.read_step, future, n, total)
    return
  if not res: # EOF
    future.set_result(bytes(self.rbuffer))
    return
  self.rbuffer.extend(res)
  self.loop.call_soon(self.read_step, future, self.BLOCK_SIZE, total)

def read(self, n=-1):
  future = asyncio.Future(loop=self.loop)

  self.rbuffer.clear()
  self.loop.call_soon(self.read_step, future, min(self.BLOCK_SIZE, n), n)

  return future