programing

프로그램의 단일 인스턴스만 실행 중인지 확인

lastcode 2023. 7. 16. 13:34
반응형

프로그램의 단일 인스턴스만 실행 중인지 확인

프로그램의 인스턴스를 하나만 실행할 수 있는 파이썬 방식이 있습니까?

제가 생각해낸 유일한 합리적인 해결책은 그것을 어떤 포트에서 서버로 실행하려고 시도하다가 같은 포트에 바인딩하려는 두 번째 프로그램이 실패하는 것입니다.하지만 좋은 생각은 아닙니다. 이것보다 더 가벼운 것이 있을까요?

(프로그램이 때때로 실패할 것으로 예상된다는 점을 고려하십시오. 즉, segfault - "파일 잠금"과 같은 기능이 작동하지 않습니다.)

다음 코드가 작업을 수행해야 하며 교차 플랫폼이며 Python 2.4-3.2에서 실행됩니다.Windows, OS X 및 Linux에서 테스트했습니다.

from tendo import singleton
me = singleton.SingleInstance() # will sys.exit(-1) if other instance is running

최신 코드 버전은 singleton.py 에서 확인할 수 있습니다.여기에 버그를 신고해 주세요.

다음 방법 중 하나를 사용하여 tend를 설치할 수 있습니다.

  • easy_install tendo
  • pip install tendo
  • http://pypi.python.org/pypi/tendo 에서 수동으로 다운로드합니다.

zgoda다른 질문에서 찾을 수 있는 단순한 교차 플랫폼 솔루션:

import fcntl
import os
import sys

def instance_already_running(label="default"):
    """
    Detect if an an instance with the label is already running, globally
    at the operating system level.

    Using `os.open` ensures that the file pointer won't be closed
    by Python's garbage collector after the function's scope is exited.

    The lock will be released when the program exits, or could be
    released if the file pointer were closed.
    """

    lock_file_pointer = os.open(f"/tmp/instance_{label}.lock", os.O_WRONLY)

    try:
        fcntl.lockf(lock_file_pointer, fcntl.LOCK_EX | fcntl.LOCK_NB)
        already_running = False
    except IOError:
        already_running = True

    return already_running

S처럼 많이.로트의 제안이지만, 코드를 가지고 있습니다.

이 코드는 Linux 전용입니다.그것은 '추상적인' UNIX 도메인 소켓을 사용하지만, 간단하고 오래된 잠금 파일을 남기지 않습니다.특별히 예약된 TCP 포트가 필요하지 않기 때문에 위의 솔루션보다 선호합니다.

try:
    import socket
    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    ## Create an abstract socket, by prefixing it with null. 
    s.bind( '\0postconnect_gateway_notify_lock') 
except socket.error as e:
    error_code = e.args[0]
    error_string = e.args[1]
    print "Process already running (%d:%s ). Exiting" % ( error_code, error_string) 
    sys.exit (0) 

문자열 유문 열자postconnect_gateway_notify_lock단일 인스턴스를 실행해야 하는 여러 프로그램을 허용하도록 변경할 수 있습니다.

그것이 충분히 파이썬적인지는 모르겠지만 정의된 포트에서 청취하는 것은 모든 주요 플랫폼에서 작동하고 프로그램 충돌에 문제가 없기 때문에 꽤 널리 사용되는 솔루션입니다.

포트 수신 대기의 또 다른 이점은 실행 중인 인스턴스에 명령을 전송할 수 있다는 것입니다.예를 들어 사용자가 프로그램을 다시 시작할 때 실행 중인 인스턴스에 다른 창을 열라는 명령을 보낼 수 있습니다(예: Firefox가 수행하는 작업).하지만 그들이 TCP 포트나 명명된 파이프 같은 것을 사용하는지는 모릅니다.

이전에 파이썬을 작성한 적은 없지만, 이것은 크론드에 의해 두 번 이상 시작되는 것을 방지하기 위해 방금 체크포인트에 구현한 것입니다.

import os
import sys
import fcntl
fh=0
def run_once():
    global fh
    fh=open(os.path.realpath(__file__),'r')
    try:
        fcntl.flock(fh,fcntl.LOCK_EX|fcntl.LOCK_NB)
    except:
        os._exit(0)

run_once()

Slava-N의 제안을 다른 이슈(http://stackoverflow.com/questions/2959474) 에 올린 후에 찾았습니다.이를 함수라고 하며 실행 중인 스크립트 파일(pid 파일이 아님)을 잠그고 스크립트가 종료될 때까지 잠금을 유지합니다(정상 또는 오류).

pid 파일을 사용합니다."/path/to/pidfile"이라는 알려진 위치가 있고 시작할 때 다음과 같은 작업을 수행합니다(부분적으로 나는 프리커피라서 그렇게 열심히 일하고 싶지 않기 때문에 유사 코드).

import os, os.path
pidfilePath = """/path/to/pidfile"""
if os.path.exists(pidfilePath):
   pidfile = open(pidfilePath,"r")
   pidString = pidfile.read()
   if <pidString is equal to os.getpid()>:
      # something is real weird
      Sys.exit(BADCODE)
   else:
      <use ps or pidof to see if the process with pid pidString is still running>
      if  <process with pid == 'pidString' is still running>:
          Sys.exit(ALREADAYRUNNING)
      else:
          # the previous server must have crashed
          <log server had crashed>
          <reopen pidfilePath for writing>
          pidfile.write(os.getpid())
else:
    <open pidfilePath for writing>
    pidfile.write(os.getpid())

즉, pid 파일이 존재하는지 확인하고, 존재하지 않으면 해당 파일에 pid를 기록합니다.pid 파일이 있으면 pid가 실행 중인 프로세스의 pid인지 확인합니다. 그렇다면 다른 라이브 프로세스가 실행 중이므로 그냥 종료하십시오.그렇지 않으면 이전 프로세스가 충돌했으므로 로그에 기록한 다음 이전 프로세스 대신 자신의 PID를 파일에 기록합니다.그럼 계속해주세요.

윈도우에서 이에 대한 가장 좋은 해결책은 @zgoda에서 제안한 대로 뮤텍스를 사용하는 것입니다.

import win32event
import win32api
from winerror import ERROR_ALREADY_EXISTS

mutex = win32event.CreateMutex(None, False, 'name')
last_error = win32api.GetLastError()

if last_error == ERROR_ALREADY_EXISTS:
   print("App instance already running")

일부 답변은 다음과 같습니다.fctnl(@package에서도 ) 수 (@ included됨포함도)와 같은 패키지를 사용하여 파이썬 앱을 pyinstaller정적 가져오기를 수행하면 오류가 발생합니다.

을 사용하여 또한잠파사다만음듭다니을여용하방을 합니다.read-only데이터베이스 파일 문제(에서 경험한 문제)sqlite3).

다음은 최종적으로 Windows 전용 솔루션입니다.다음을 'onlyone.py '이라고 하는 모듈에 넣으십시오.이 모듈을 __ 메인 __ 파이썬 스크립트 파일에 직접 포함합니다.

import win32event, win32api, winerror, time, sys, os
main_path = os.path.abspath(sys.modules['__main__'].__file__).replace("\\", "/")

first = True
while True:
        mutex = win32event.CreateMutex(None, False, main_path + "_{<paste YOUR GUID HERE>}")
        if win32api.GetLastError() == 0:
            break
        win32api.CloseHandle(mutex)
        if first:
            print "Another instance of %s running, please wait for completion" % main_path
            first = False
        time.sleep(1)

설명.

코드는 스크립트의 전체 경로에서 파생된 이름으로 뮤텍스를 만들려고 시도합니다.실제 파일 시스템과의 혼동을 방지하기 위해 슬래시를 사용합니다.

이점

  • 구성이나 '매직' 식별자가 필요하지 않으므로 필요한 만큼 다양한 스크립트에서 사용합니다.
  • 오래된 파일은 남아있지 않습니다. 뮤텍스도 함께 사라집니다.
  • 대기 시 유용한 메시지 인쇄

이것은 효과가 있을 것입니다.

  1. 알려진 위치에 PID 파일을 만들려고 합니다.실패하면 다른 사용자가 파일을 잠그면 완료됩니다.

  2. 정상적으로 완료되면 다른 사용자가 덮어쓸 수 있도록 PID 파일을 닫았다가 제거합니다.

프로그램이 충돌하더라도 PID 파일을 제거하는 셸 스크립트로 프로그램을 래핑할 수 있습니다.

또한 프로그램이 중단되면 PID 파일을 사용하여 프로그램을 종료할 수 있습니다.

응용 프로그램에 wxPython을 사용하는 사용자는 누구나 이 기능을 사용할 수 있습니다.wx.SingleInstanceChecker 여기에 문서화되어 있습니다.

으로 개적으사하클는래스의 합니다.wx.App wx.SingleInstanceChecker 와리즈턴을 반환합니다.FalseOnInit() 경우:

import wx

class SingleApp(wx.App):
    """
    class that extends wx.App and only permits a single running instance.
    """

    def OnInit(self):
        """
        wx.App init function that returns False if the app is already running.
        """
        self.name = "SingleApp-%s".format(wx.GetUserId())
        self.instance = wx.SingleInstanceChecker(self.name)
        if self.instance.IsAnotherRunning():
            wx.MessageBox(
                "An instance of the application is already running", 
                "Error", 
                 wx.OK | wx.ICON_WARNING
            )
            return False
        return True

이것은 다음을 위한 간단한 드롭인 교체입니다.wx.App여러 인스턴스를 금지합니다.사용 방법은 간단히 교체wx.App와 함께SingleApp다음과 같은 코드로:

app = SingleApp(redirect=False)
frame = wx.Frame(None, wx.ID_ANY, "Hello World")
frame.Show(True)
app.MainLoop()

잠금 파일을 사용하는 것은 유닉스에서 상당히 일반적인 접근 방식입니다.충돌이 발생하면 수동으로 청소해야 합니다.파일에 PID를 저장할 수 있으며, 시작 시 이 PID에 프로세스가 있는지 확인하고, 없으면 잠금 파일을 재정의할 수 있습니다. (단, 읽기-파일-체크-pid-rewrite-파일 주변의 잠금도 필요합니다.)당신은 os-package에서 pid를 얻고 확인하는 데 필요한 것을 찾을 수 있을 것입니다.주어진 pid에 공정이 있는지 확인하는 일반적인 방법은 치명적이지 않은 신호를 보내는 것입니다.

다른 대안은 이것을 무리 또는 포식스 세마포어와 결합하는 것일 수 있습니다.

saua가 제안한 것처럼 네트워크 소켓을 여는 것이 아마도 가장 쉽고 휴대하기 쉬울 것입니다.

로베르토 로사리오의 답변을 바탕으로 다음과 같은 기능을 생각해 냈습니다.

SOCKET = None
def run_single_instance(uniq_name):
    try:
        import socket
        global SOCKET
        SOCKET = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        ## Create an abstract socket, by prefixing it with null.
        # this relies on a feature only in linux, when current process quits, the
        # socket will be deleted.
        SOCKET.bind('\0' + uniq_name)
        return True
    except socket.error as e:
        return False

우리는 글로벌을 정의해야 합니다.SOCKET전체 프로세스가 종료될 때만 가비지를 수집하므로 사용할 수 있습니다.함수에 로컬 변수를 선언하면 함수가 종료된 후 범위를 벗어나 소켓이 삭제됩니다.

모든 공은 로베르토 로사리오에게 돌아가야 합니다. 왜냐하면 저는 그의 코드에 대해 명확하고 자세히 설명할 뿐이기 때문입니다.https://troydhanson.github.io/network/Unix_domain_sockets.html 의 다음 인용문에서 설명하는 것처럼 이 코드는 Linux에서만 작동합니다.

Linux에는 특수 기능이 있습니다. UNIX 도메인 소켓의 경로 이름이 null 바이트 \0으로 시작하면 이름이 파일 시스템에 매핑되지 않습니다.따라서 파일 시스템의 다른 이름과 충돌하지 않습니다.또한 서버가 추상 네임스페이스에서 UNIX 도메인 수신 소켓을 닫으면 파일이 삭제됩니다. 일반 UNIX 도메인 소켓의 경우 서버가 해당 소켓을 닫은 후에도 파일이 유지됩니다.

답변이 늦었지만 창에 대해서는 다음을 사용할 수 있습니다.

from win32event import CreateMutex
from win32api import CloseHandle, GetLastError
from winerror import ERROR_ALREADY_EXISTS
import sys

class singleinstance:
    """ Limits application to single instance """

    def __init__(self):
        self.mutexname = "testmutex_{D0E858DF-985E-4907-B7FB-8D732C3FC3B9}"
        self.mutex = CreateMutex(None, False, self.mutexname)
        self.lasterror = GetLastError()
    
    def alreadyrunning(self):
        return (self.lasterror == ERROR_ALREADY_EXISTS)
        
    def __del__(self):
        if self.mutex:
            CloseHandle(self.mutex)

사용.

# do this at beginnig of your application
myapp = singleinstance()

# check is another instance of same program running
if myapp.alreadyrunning():
    print ("Another instance of this program is already running")
    sys.exit(1)

제가 새로운 사용자이고 Stack Overflow가 아직 투표를 못하게 해서 답변으로 글을 올립니다.

Sorin Sbarnea의 솔루션은 OS X, Linux 및 Windows에서 작동하며 감사합니다.

그러나 tempfile.gettempdir()는 OS X 및 Windows에서는 한 가지 방식으로 작동하고 다른 일부/다중/다중(?)*nixes에서는 다른 방식으로 작동합니다(OS X도 Unix!라는 사실은 무시).그 차이는 이 코드에 중요합니다.

OS X와 Windows에는 사용자별 임시 디렉토리가 있으므로 한 사용자가 만든 임시 파일은 다른 사용자에게 표시되지 않습니다.이와 대조적으로 많은 *nix 버전(Ubuntu 9, RHEL 5, OpenSolaris 2008 및 FreeBSD 8 테스트)에서 tempdir는 모든 사용자에 대해 /tmp입니다.

즉, 다중 사용자 컴퓨터에서 잠금 파일이 생성되면 /tmp로 생성되고 잠금 파일을 처음 생성한 사용자만 응용 프로그램을 실행할 수 있습니다.

가능한 해결책은 잠금 파일의 이름에 현재 사용자 이름을 포함시키는 것입니다.

포트를 잡는 OP의 솔루션이 다중 사용자 시스템에서도 잘못 작동할 것이라는 점에 주목할 필요가 있습니다.

다음은 Python 3.7.9를 사용하여 Windows Server 2016 및 Ubuntu 20.04에서 테스트한 교차 플랫폼 예입니다.

import os

class SingleInstanceChecker:
    def __init__(self, id):
        if isWin():
            ensure_win32api()
            self.mutexname = id
            self.lock = win32event.CreateMutex(None, False, self.mutexname)
            self.running = (win32api.GetLastError() == winerror.ERROR_ALREADY_EXISTS)

        else:
            ensure_fcntl()
            self.lock = open(f"/tmp/isnstance_{id}.lock", 'wb')
            try:
                fcntl.lockf(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
                self.running = False
            except IOError:
                self.running = True


    def already_running(self):
        return self.running
        
    def __del__(self):
        if self.lock:
            try:
                if isWin():
                    win32api.CloseHandle(self.lock)
                else:
                    os.close(self.lock)
            except Exception as ex:
                pass

# ---------------------------------------
# Utility Functions
# Dynamically load win32api on demand
# Install with: pip install pywin32
win32api=winerror=win32event=None
def ensure_win32api():
    global win32api,winerror,win32event
    if win32api is None:
        import win32api
        import winerror
        import win32event


# Dynamically load fcntl on demand
# Install with: pip install fcntl
fcntl=None
def ensure_fcntl():
    global fcntl
    if fcntl is None:
        import fcntl


def isWin():
    return (os.name == 'nt')
# ---------------------------------------

다음은 사용 중인 항목입니다.

import time, sys

def main(argv):
    _timeout = 10
    print("main() called. sleeping for %s seconds" % _timeout)
    time.sleep(_timeout)
    print("DONE")


if __name__ == '__main__':
    SCR_NAME = "my_script"
    sic = SingleInstanceChecker(SCR_NAME)
    if sic.already_running():
        print("An instance of {} is already running.".format(SCR_NAME))
        sys.exit(1)
    else:
        main(sys.argv[1:])

사용합니다single_process나의 젠투;

pip install single_process

:

from single_process import single_process

@single_process
def main():
    print 1

if __name__ == "__main__":
    main()   

참조: https://pypi.python.org/pypi/single_process/

저는 파일 시스템을 사용하지 않고도 프로세스 그룹을 사용하는 좋은 POSIXy 솔루션이 있어야 한다고 계속 의심하고 있지만, 정확히 파악할 수 없습니다.다음과 같은 것:

시작할 때 프로세스는 특정 그룹의 모든 프로세스에 'kill -0'을 보냅니다.이러한 프로세스가 있으면 종료됩니다.그런 다음 그룹에 가입합니다.다른 프로세스에서는 해당 그룹을 사용하지 않습니다.

그러나 이는 경쟁 조건을 가지고 있습니다. 즉, 여러 프로세스가 동시에 이 작업을 수행할 수 있으며, 그룹에 가입하여 동시에 실행될 수 있습니다.수밀 상태로 만들기 위해 일종의 뮤텍스를 추가하면 더 이상 공정 그룹이 필요하지 않습니다.

이는 프로세스가 cron에 의해 1분 또는 1시간에 한 번만 시작되는 경우에 허용될 수 있지만, 당신이 원하지 않는 날에 정확히 잘못될까 봐 약간 불안합니다.

누군가가 개선할 수 없는 한, 결국 이것은 그다지 좋은 해결책이 아니라고 생각합니다.

저는 지난주에 이 정확한 문제에 부딪혔고, 몇 가지 좋은 해결책을 찾았지만, 매우 간단하고 깨끗한 파이썬 패키지를 만들기로 결정하고 PyPI에 업로드했습니다.모든 문자열 리소스 이름을 잠글 수 있다는 점에서 텐도와 다릅니다.당신이 확실히 잠글 수는 있지만,__file__동일한 효과를 얻기 위해.

대상: 설치대:pip install quicklock

사용법은 매우 간단합니다.

[nate@Nates-MacBook-Pro-3 ~/live] python
Python 2.7.6 (default, Sep  9 2014, 15:04:36)
[GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.39)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> from quicklock import singleton
>>> # Let's create a lock so that only one instance of a script will run
...
>>> singleton('hello world')
>>>
>>> # Let's try to do that again, this should fail
...
>>> singleton('hello world')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/nate/live/gallery/env/lib/python2.7/site-packages/quicklock/quicklock.py", line 47, in singleton
    raise RuntimeError('Resource <{}> is currently locked by <Process {}: "{}">'.format(resource, other_process.pid, other_process.name()))
RuntimeError: Resource <hello world> is currently locked by <Process 24801: "python">
>>>
>>> # But if we quit this process, we release the lock automatically
...
>>> ^D
[nate@Nates-MacBook-Pro-3 ~/live] python
Python 2.7.6 (default, Sep  9 2014, 15:04:36)
[GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.39)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> from quicklock import singleton
>>> singleton('hello world')
>>>
>>> # No exception was thrown, we own 'hello world'!

확인: https://pypi.python.org/pypi/quicklock

Linux 예제

이 방법은 응용프로그램을 닫은 후 자동으로 삭제되는 임시 파일을 만드는 것을 기반으로 합니다.프로그램 실행은 파일의 존재를 확인합니다. 파일이 존재하면 프로그램이 닫힙니다. 그렇지 않으면 파일을 만들고 프로그램의 실행을 계속합니다.

from tempfile import *
import time
import os
import sys


f = NamedTemporaryFile( prefix='lock01_', delete=True) if not [f  for f in     os.listdir('/tmp') if f.find('lock01_')!=-1] else sys.exit()

YOUR CODE COMES HERE

에서는 리눅스에 문의할 .pgrep -a인스턴스 수에 대한 스크립트는 프로세스 목록에 있습니다(옵션 -a에는 전체 명령줄 문자열이 표시됨).

import os
import sys
import subprocess

procOut = subprocess.check_output( "/bin/pgrep -u $UID -a python", shell=True, 
                                   executable="/bin/bash", universal_newlines=True)

if procOut.count( os.path.basename(__file__)) > 1 :        
    sys.exit( ("found another instance of >{}<, quitting."
              ).format( os.path.basename(__file__)))

제다한을 합니다.-u $UID제한이 모든 사용자에게 적용되어야 하는 경우.고지 사항: a) 스크립트의 (기본) 이름이 고유하다고 가정하고, b) 경합 조건이 있을 수 있습니다.

다음은 컨텍스트 매니저와 memcached가 있는 django의 좋은 예입니다: https://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html

서로 다른 호스트에서 동시 작업을 보호하는 데 사용할 수 있습니다.여러 작업을 관리하는 데 사용할 수 있습니다.단순한 파이썬 스크립트에 대해서도 변경할 수 있습니다.

위 코드에 대한 나의 수정 사항은 다음과 같습니다.

import time
from contextlib import contextmanager
from django.core.cache import cache


@contextmanager
def memcache_lock(lock_key, lock_value, lock_expire):
    timeout_at = time.monotonic() + lock_expire - 3

    # cache.add fails if the key already exists
    status = cache.add(lock_key, lock_value, lock_expire)
    try:
        yield status
    finally:
        # memcache delete is very slow, but we have to use it to take
        # advantage of using add() for atomic locking
        if time.monotonic() < timeout_at and status:
            # don't release the lock if we exceeded the timeout
            # to lessen the chance of releasing an expired lock owned by someone else
            # also don't release the lock if we didn't acquire it
            cache.delete(lock_key)


LOCK_EXPIRE = 60 * 10  # Lock expires in 10 minutes


def main():
    lock_name, lock_value = "lock_1", "locked"
    with memcache_lock(lock_name, lock_value, LOCK_EXPIRE) as acquired:
        if acquired:
            # single instance code here:
            pass


if __name__ == "__main__":
    main()

다음은 컨텍스트 관리자를 사용하여 임시 잠금 파일을 만드는 교차 플랫폼 구현입니다.

여러 작업을 관리하는 데 사용할 수 있습니다.

import os
from contextlib import contextmanager
from time import sleep


class ExceptionTaskInProgress(Exception):
    pass


# Context manager for suppressing exceptions
class SuppressException:
    def __init__(self):
        pass

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return True


# Context manager for task
class TaskSingleInstance:
    def __init__(self, task_name, lock_path):
        self.task_name = task_name
        self.lock_path = lock_path
        self.lock_filename = os.path.join(self.lock_path, self.task_name + ".lock")

        if os.path.exists(self.lock_filename):
            raise ExceptionTaskInProgress("Resource already in use")

    def __enter__(self):
        self.fl = open(self.lock_filename, "w")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.fl.close()
        os.unlink(self.lock_filename)


# Here the task is silently interrupted
# if it is already running on another instance.
def main1():
    task_name = "task1"
    tmp_filename_path = "."
    with SuppressException():
        with TaskSingleInstance(task_name, tmp_filename_path):
            print("The task `{}` has started.".format(task_name))
            # The single task instance code is here.
            sleep(5)
            print("The task `{}` has completed.".format(task_name))


# Here the task is interrupted with a message
# if it is already running in another instance.
def main2():
    task_name = "task1"
    tmp_filename_path = "."
    try:
        with TaskSingleInstance(task_name, tmp_filename_path):
            print("The task `{}` has started.".format(task_name))
            # The single task instance code is here.
            sleep(5)
            print("Task `{}` completed.".format(task_name))
    except ExceptionTaskInProgress as ex:
        print("The task `{}` is already running.".format(task_name))


if __name__ == "__main__":
    main1()
    main2()

잠금 전용 이름을 가진 파일을 만듭니다.파이의

import os
import sys
import atexit
import hashlib


@atexit.register # clean up at exit 
def cleanup():
    try:
        if config.lock_file:
            config.lock_file.close()
        if config.fname:
            os.remove(config.fname)
    except Exception:
        pass


config = sys.modules[__name__] # this allows us to share variables with the main script
config.file = None
config.fname = None
config.lock_file = None
config.maxinstances = 1


def configure_lock(
    maxinstances: int = 1,
    message: str | None = None,
    file: str | None = None,
) -> None:
    """
    Configures a lock file for a given file path and maximum number of instances.

    Args:
        maxinstances (int, optional): The maximum number of instances allowed to access the file. Defaults to 1.
        message (str, optional): The message to print if the maximum number of instances is reached. Defaults to None.
        file (str, optional): The file path to configure the lock file for. Defaults to None.


    Returns:
        None

    Raises:
        None
    """
    if not file: # if not file is passed, we get the calling filename from the frame
        f = sys._getframe(1)
        dct = f.f_globals
        file = dct.get("__file__", "")
    config.file = os.path.normpath(file)
    config.maxinstances = int(maxinstances)

    for inst in range(config.maxinstances):
        try:
            hash = hashlib.sha256((config.file + f"{inst}").encode("utf-8", "ignore")) # unique name to make sure other that it doesn't interfere with other py files using this function 
            config.fname = hash.digest().hex() + ".locfi"
            tmpf = os.path.join(os.environ.get("TMP"), config.fname)
            if os.path.exists(tmpf):
                os.remove(tmpf)
            config.lock_file = os.open(tmpf, os.O_CREAT | os.O_EXCL)
            break
        except Exception as fe:
            if inst + 1 == config.maxinstances:
                if message:
                    print(message)
                try:
                    sys.exit(1)
                finally:
                    os._exit(1) # just to make sure :) 
            else:
                continue

스크립트로 가져오기:

import sys
from time import sleep

from lockexclusive import configure_lock

# it can be used like this:
# configure_lock(maxinstances=1, message="More than one instance running",file=sys.argv[0])

# or without the file argument:
configure_lock(maxinstances=1, message="More than one instance running")

sleep(100)
import sys,os

# start program
try:  # (1)
    os.unlink('lock')  # (2)
    fd=os.open("lock", os.O_CREAT|os.O_EXCL) # (3)  
except: 
    try: fd=os.open("lock", os.O_CREAT|os.O_EXCL) # (4) 
    except:  
        print "Another Program running !.."  # (5)
        sys.exit()  

# your program  ...
# ...

# exit program
try: os.close(fd)  # (6)
except: pass
try: os.unlink('lock')  
except: pass
sys.exit()  

언급URL : https://stackoverflow.com/questions/380870/make-sure-only-a-single-instance-of-a-program-is-running

반응형