Upewnij się, że działa tylko jedno wystąpienie programu

Czy istnieje sposób w Pythonie, aby uruchomić tylko jedną instancję programu?

Jedynym rozsądnym rozwiązaniem, które wymyśliłem, jest próba uruchomienia go jako serwera na jakimś porcie, a następnie drugi program próbujący połączyć się z tym samym portem - kończy się niepowodzeniem. Ale to naprawdę nie jest dobry pomysł, może jest coś lżejszego niż to?

(Weź pod uwagę, że czasami oczekuje się, że program zawiedzie, np. Segfault - więc rzeczy takie jak „lock file” nie będą działać)

Poniższy kod powinien wykonać zadanie, jest wieloplatformowy i działa na Pythonie 2.4-3.2. Przetestowałem to na Windowsie, OS X i Linuksie.

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

Najnowsza wersja kodu jest dostępna na singleton.py . Prosimy o zgłaszanie błędów tutaj .

Możesz zainstalować program za pomocą jednej z następujących metod:

• easy_install tendo
• pip install tendo
• ręcznie, pobierając go z http://pypi.python.org/pypi/tendo

Prosty, wieloplatformowe rozwiązanie, znalezione w innym pytaniu przez 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

Bardzo podobne do sugestii S. Lotta, ale z kodem.

Ten kod jest specyficzny dla systemu Linux. Używa „abstrakcyjnych” gniazd domeny UNIX, ale jest prosty i nie pozostawia starych plików blokady. Wolę to od rozwiązania powyżej, ponieważ nie wymaga specjalnie zarezerwowanego portu 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) 

Unikalny ciąg postconnect_gateway_notify_lockmożna zmienić, aby zezwolić na wiele programów, które wymagają wymuszenia jednej instancji.

Nie wiem, czy jest wystarczająco pythonowy, ale w świecie Javy nasłuchiwanie na określonym porcie jest dość szeroko stosowanym rozwiązaniem, ponieważ działa na wszystkich głównych platformach i nie ma żadnych problemów z zawieszaniem się programów.

Kolejną zaletą nasłuchiwania portu jest możliwość wysłania polecenia do działającej instancji. Na przykład, gdy użytkownicy uruchamiają program po raz drugi, możesz wysłać działającej instancji polecenie, aby nakazała jej otwarcie innego okna (na przykład to robi Firefox. Nie wiem, czy używają portów TCP, nazwanych potoków lub coś w tym rodzaju '').

Nigdy wcześniej nie pisałem Pythona, ale właśnie to zaimplementowałem w mycheckpoint, aby zapobiec uruchamianiu go dwa lub więcej razy przez crond:

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()

Znalazłem sugestię Slava-N po opublikowaniu tego w innym numerze (http://stackoverflow.com/questions/2959474). Ta jest wywoływana jako funkcja, blokuje plik wykonywanych skryptów (nie plik pid) i utrzymuje blokadę do zakończenia skryptu (normalny lub błąd).

Użyj pliku pid. Masz jakąś znaną lokalizację, „/ path / to / pidfile” i podczas uruchamiania robisz coś takiego (częściowo pseudokod, ponieważ przygotowuję kawę i nie chcę tak ciężko pracować):

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())

Innymi słowy, sprawdzasz, czy istnieje plik pidfile; jeśli nie, zapisz swój pid do tego pliku. Jeśli plik pid istnieje, sprawdź, czy pid jest numerem pid działającego procesu; jeśli tak, to masz uruchomiony inny proces na żywo, więc po prostu zamknij. Jeśli nie, to poprzedni proces się zawiesił, więc zaloguj się, a następnie zapisz swój własny pid w pliku zamiast starego. Następnie kontynuuj.

Znalazłeś już odpowiedź na podobne pytanie w innym wątku, więc ze względu na kompletność zobacz, jak osiągnąć to samo w systemie Windows bez nazwy mutex.

http://code.activestate.com/recipes/474070/

To może zadziałać.

1. Spróbuj utworzyć plik PID w znanej lokalizacji. Jeśli ci się nie uda, ktoś zablokował plik - gotowe.

2. Kiedy skończysz normalnie, zamknij i usuń plik PID, aby ktoś inny mógł go zastąpić.

Możesz opakować swój program w skrypt powłoki, który usuwa plik PID nawet w przypadku awarii programu.

Możesz także użyć pliku PID do zabicia programu, jeśli się zawiesi.

Używanie pliku blokady jest dość powszechnym podejściem na Uniksie. Jeśli się zawiesza, musisz wyczyścić ręcznie. Możesz zapisać PID w pliku i przy uruchomieniu sprawdzić, czy istnieje proces z tym PID, nadpisując plik blokady, jeśli nie. (Jednak potrzebujesz również blokady wokół pliku read-file-check-pid-rewrite-file). W pakiecie os znajdziesz to, czego potrzebujesz do pobrania i sprawdzenia pid . Powszechnym sposobem sprawdzenia, czy istnieje proces z danym pidem, jest wysłanie mu nie-fatalnego sygnału.

Inną alternatywą mogłoby być połączenie tego z semaforami typu flock lub Posix.

Otwarcie gniazda sieciowego, zgodnie z propozycją saua, byłoby prawdopodobnie najłatwiejsze i najbardziej przenośne.

Każdy, kto używa wxPython w swojej aplikacji, może skorzystać z funkcji wx.SingleInstanceChecker opisanej tutaj .

Osobiście używam podklasy, wx.Appktóra korzysta z wx.SingleInstanceCheckeri wraca Falsez, OnInit()jeśli istnieje już instancja aplikacji, która jest już wykonywana w następujący sposób:

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

Jest to prosty zamiennik typu drop-in, wx.Appktóry uniemożliwia wiele wystąpień. Aby z niego skorzystać wystarczy wymienić wx.Appze SingleAppw kodzie tak:

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

Oto moje ostateczne rozwiązanie tylko dla systemu Windows. Wstaw poniższy kod do modułu, być może o nazwie „onlyone.py” lub jakikolwiek inny. Dołącz ten moduł bezpośrednio do swojego __ głównego __ pliku skryptu języka Python.

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)

Wyjaśnienie

Kod próbuje utworzyć muteks o nazwie pochodzącej z pełnej ścieżki do skryptu. Używamy ukośników, aby uniknąć potencjalnego pomylenia z rzeczywistym systemem plików.

Zalety

• Nie są potrzebne żadne identyfikatory konfiguracji ani „magiczne” - używaj ich w tylu różnych skryptach, ile potrzeba.
• Nie ma przestarzałych plików, mutex umiera razem z tobą.
• Podczas oczekiwania drukuje pomocny komunikat

Najlepszym rozwiązaniem w przypadku okien jest użycie muteksów, zgodnie z sugestią @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")

Niektóre odpowiedzi używają fctnl(zawarte również w pakiecie @sorin tendo), które nie jest dostępne w systemie Windows i jeśli spróbujesz zamrozić aplikację Pythona za pomocą pakietu, takiego jak pyinstallerimport statyczny, generuje błąd.

Ponadto użycie metody pliku blokującego stwarza read-onlyproblem z plikami bazy danych (doświadczony w przypadku sqlite3).

Publikuję to jako odpowiedź, ponieważ jestem nowym użytkownikiem, a Stack Overflow nie pozwala mi jeszcze głosować.

Rozwiązanie Sorin Sbarnea działa dla mnie pod OS X, Linux i Windows i jestem za to wdzięczny.

Jednak tempfile.gettempdir () zachowuje się w jeden sposób pod OS X i Windows, a w inny pod niektórymi / many / all (?) * Nixami (ignorując fakt, że OS X jest również Uniksem!). Różnica jest ważna w tym kodzie.

Systemy OS X i Windows mają katalogi tymczasowe specyficzne dla użytkownika, więc plik tymczasowy utworzony przez jednego użytkownika nie jest widoczny dla innego użytkownika. Z kolei w wielu wersjach * nix (testowałem Ubuntu 9, RHEL 5, OpenSolaris 2008 i FreeBSD 8), katalog tymczasowy to / tmp dla wszystkich użytkowników.

Oznacza to, że gdy plik blokujący jest tworzony na komputerze z wieloma użytkownikami, jest tworzony w / tmp i tylko użytkownik, który utworzył plik blokujący po raz pierwszy, będzie mógł uruchomić aplikację.

Możliwym rozwiązaniem jest osadzenie aktualnej nazwy użytkownika w nazwie pliku blokady.

Warto zauważyć, że rozwiązanie OP polegające na przechwytywaniu portu będzie również źle działać na maszynie wielodostępnej.

Używam single_processna moim gentoo;

pip install single_process

przykład :

from single_process import single_process

@single_process
def main():
    print 1

if __name__ == "__main__":
    main()   

patrz: https://pypi.python.org/pypi/single_process/1.0

Podejrzewam, że powinno być dobre rozwiązanie POSIXy wykorzystujące grupy procesów, bez konieczności uderzania w system plików, ale nie mogę tego całkiem ustalić. Coś jak:

Podczas uruchamiania proces wysyła komunikat „kill -0” do wszystkich procesów w określonej grupie. Jeśli istnieją takie procesy, kończy się. Następnie dołącza do grupy. Żadne inne procesy nie używają tej grupy.

Jednak jest to warunek wyścigu - wiele procesów może to zrobić dokładnie w tym samym czasie i wszystkie ostatecznie dołączają do grupy i działają jednocześnie. Do czasu dodania jakiegoś muteksu, aby uczynić go wodoszczelnym, nie potrzebujesz już grup procesów.

Może to być akceptowalne, jeśli twój proces jest uruchamiany tylko przez crona, raz na minutę lub co godzinę, ale denerwuje mnie trochę, że pójdzie źle dokładnie w dniu, w którym tego nie chcesz.

Wydaje mi się, że to jednak nie jest zbyt dobre rozwiązanie, chyba że ktoś może to poprawić?

W zeszłym tygodniu napotkałem dokładnie ten problem i chociaż znalazłem kilka dobrych rozwiązań, zdecydowałem się stworzyć bardzo prosty i czysty pakiet Pythona i załadowałem go do PyPI. Różni się od tendo tym, że może zablokować dowolną nazwę zasobu ciągu. Chociaż z pewnością można by było zablokować, __file__aby osiągnąć ten sam efekt.

Zainstaluj za pomocą: pip install quicklock

Korzystanie z niego jest niezwykle proste:

[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'!

Spójrz: https://pypi.python.org/pypi/quicklock

Opierając się na odpowiedzi Roberto Rosario, wymyślam następującą funkcję:

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

Musimy zdefiniować global SOCKETvaiable, ponieważ będzie on zbierany jako śmieci dopiero po zakończeniu całego procesu. Jeśli zadeklarujemy zmienną lokalną w funkcji, wyjdzie ona poza zakres po zakończeniu funkcji, a tym samym gniazdo zostanie usunięte.

Wszystko to zasługa Roberto Rosario, ponieważ ja tylko wyjaśniam i rozwijam jego kod. Ten kod będzie działał tylko w Linuksie, jak wyjaśnia poniższy cytowany tekst z https://troydhanson.github.io/network/Unix_domain_sockets.html :

Linux ma specjalną cechę: jeśli ścieżka dostępu do gniazda domeny UNIX zaczyna się od bajtu zerowego \ 0, jego nazwa nie jest odwzorowywana w systemie plików. Dlatego nie będzie kolidował z innymi nazwami w systemie plików. Ponadto, gdy serwer zamyka swoje gniazdo nasłuchujące domeny UNIX w abstrakcyjnej przestrzeni nazw, jego plik jest usuwany; ze zwykłymi gniazdami domeny UNIX, plik pozostaje po zamknięciu przez serwer.

przykład linux

Ta metoda polega na utworzeniu pliku tymczasowego automatycznie usuwanego po zamknięciu aplikacji. po uruchomieniu programu weryfikujemy istnienie pliku; jeśli plik istnieje (oczekuje na wykonanie), program jest zamykany; w przeciwnym razie tworzy plik i kontynuuje wykonywanie programu.

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

W systemie Linux można również zapytać pgrep -ao liczbę instancji, skrypt znajduje się na liście procesów (opcja -a ujawnia pełny ciąg wiersza poleceń). Na przykład

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__)))

Usuń, -u $UIDjeśli ograniczenie ma dotyczyć wszystkich użytkowników. Zastrzeżenie: a) zakłada się, że (bazowa) nazwa skryptu jest unikalna, b) mogą istnieć warunki wyścigu.

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()