排他制御

排他制御

pythonで複数プロセスの排他制御。

1. 案1 open
2. 案2 mkdir

案1 open(mode='x')

mode=x 排他的オープン。存在していた場合はエラー、存在しない場合はファイルを作成する。

import os

class test:
  failed=1
  passed=0
  bug=2
  file="lock"

  def acquire(self):
    try:
      with open(test.file,mode="x") as f:
        f.close()
    except FileExistsError:
      return test.failed

    return test.passed

  def release(self):
    try:
      os.remove(test.file)
    except OSError as e:
      pass
    except PermissionError as e:
      pass



  def doit(self):
    ret = self.acquire()
    if ret != test.passed:
      print(f"failed {ret}")
      return ret
    print("doit")
    self.release()

if __name__=='__main__':
  a=test()
  a.doit()

案2 mkdir

openの代わりにmkdirを使う以外は同じ。

import os

class test:
  failed=1
  passed=0
  bug=2
  file="lock"

  def acquire(self):
    try:
      os.mkdir(test.file)
    except FileExistsError:
      return test.failed
    except FileNotFoundError:
      rerturn test.failed

    return test.passed

  def release(self):
    os.rmdir(test.file)

  def doit(self):
    ret = self.acquire()
    if ret != test.passed:
      print(f"failed {ret}")
      return ret
    print("doit")
    self.release()

if __name__=='__main__':
  a=test()
  a.doit()

応用

3個のリソースの空きリソースを使う。一つも空いていなければエラー

import os

class test:
  failed=-1
  passed=1
  bug=-2
  noresource = 0
  files=(1,2,3)

  def __init__(self):
    self.rsc = -1

  def acquire(self): # 空きリソースを返す
    for r in test.files:
      try:
        os.mkdir(str(r))
        self.rsc = r
        return r
      except FileExistsError:
        # no resourse r
        pass
     except FileNotFoundError:
       return test.failed

    return test.noresource

  def release(self):
    if self.rsc > test.noresource:
      os.rmdir(str(self.rsc))
    self.rsc = test.failed

  def domain(self):
    ret = self.acquire()
    if ret < 0:
      print(f"failed {ret}")
      return ret

    if ret==0:
      print("no resource")
      return ret

    print(f"doit with {ret}")
    self.release()

  def doit(self):
    try:
      domain()
    except Exception as e:
      self.release()
    

if __name__=='__main__':
  a=test()
  a.doit()