Убедитесь, что запущен только один экземпляр программы

Я не знаю, достаточно ли он питонический, но в мире Java прослушивание определенного порта - довольно широко используемое решение, так как оно работает на всех основных платформах и не имеет проблем с сбоями программ.

Еще одно преимущество прослушивания порта состоит в том, что вы можете отправить команду работающему экземпляру. Например, когда пользователи запускают программу во второй раз, вы можете отправить запущенному экземпляру команду, чтобы он открыл другое окно (например, это то, что делает Firefox. Я не знаю, используют ли они TCP-порты или именованные каналы или что-то вроде этого '').

Это может сработать.

1. Попытайтесь создать файл PID в известном месте. Если вы потерпели неудачу, кто-то заблокировал файл, все готово.

2. Когда вы закончите в обычном режиме, закройте и удалите файл PID, чтобы кто-то другой мог его перезаписать.

Вы можете заключить свою программу в сценарий оболочки, который удаляет файл PID, даже если ваша программа выйдет из строя.

Вы также можете использовать файл PID, чтобы убить программу, если она зависнет.

Использование файла блокировки - довольно распространенный подход в unix. В случае сбоя необходимо очистить вручную. Вы можете сохранить PID в файле и при запуске проверить, существует ли процесс с этим PID, переопределив файл блокировки, если нет. (Однако вам также нужна блокировка вокруг файла read-file-check-pid-rewrite-file). Вы найдете то, что вам нужно для получения и проверки pid, в пакете Операционные системы. Обычный способ проверить, существует ли процесс с данным pid, - послать ему нефатальный сигнал.

Другими альтернативами могут быть комбинирование этого с семафорами flock или posix.

Открытие сетевого сокета, как предлагает saua, вероятно, будет самым простым и портативным.

Используйте файл 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 в файл вместо старого. Тогда продолжай.

Простое решение кроссплатформенный, найденное в Другой вопрос с помощью згода:

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

Очень похоже на предложение С.Лотта, но с кодом.

Следующий код должен выполнить эту работу, он кроссплатформенный и работает на 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. Пожалуйста сообщать об ошибках здесь.

Вы можете установить тендер одним из следующих способов:

• easy_install tendo
• pip install tendo
• вручную, получив его из http://pypi.python.org/pypi/tendo

Этот код специфичен для 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 можно изменить, чтобы разрешить использование нескольких программ, которым требуется один экземпляр.

Я отправляю это как ответ, потому что я новый пользователь и Stack Overflow пока не позволяет мне голосовать.

Решение Сорина Сбарнеа у меня работает под OS X, Linux и Windows, и я ему за это благодарен.

Однако tempfile.gettempdir () ведет себя одним способом под OS X и Windows, а другим - под другими some / many / all (?) * Nixes (игнорируя тот факт, что OS X также является Unix!). Для этого кода важна разница.

В OS X и Windows есть временные каталоги для конкретных пользователей, поэтому временный файл, созданный одним пользователем, не виден другому пользователю. Напротив, во многих версиях * nix (я тестировал Ubuntu 9, RHEL 5, OpenSolaris 2008 и FreeBSD 8) временный каталог - / tmp для всех пользователей.

Это означает, что когда файл блокировки создается на многопользовательской машине, он создается в / tmp, и только пользователь, который создает файл блокировки в первый раз, сможет запустить приложение.

Возможное решение - встроить текущее имя пользователя в имя файла блокировки.

Стоит отметить, что решение OP по захвату порта также будет некорректно работать на многопользовательской машине.

Никогда раньше не писал python, но это то, что я только что реализовал в mycheckpoint, чтобы предотвратить его запуск дважды или более 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()

Нашел предложение Slava-N после публикации в другом выпуске (http://stackoverflow.com/questions/2959474). Он вызывается как функция, блокирует исполняемый файл сценария (не файл pid) и поддерживает блокировку до завершения сценария (нормального или ошибочного).

Я продолжаю подозревать, что должно быть хорошее решение POSIXy, использующее группы процессов, без необходимости затрагивать файловую систему, но я не могу его понять. Что-то вроде:

При запуске ваш процесс отправляет «kill -0» всем процессам в определенной группе. Если такие процессы существуют, он завершается. Затем он присоединяется к группе. Никакие другие процессы не используют эту группу.

Однако это имеет состояние гонки - все несколько процессов могут делать это в одно и то же время, и все в конечном итоге присоединяются к группе и работают одновременно. К тому времени, когда вы добавите какой-то мьютекс, чтобы сделать его водонепроницаемым, вам больше не нужны группы процессов.

Это может быть приемлемо, если ваш процесс запускается cron только раз в минуту или каждый час, но меня немного беспокоит, что он пойдет не так именно в тот день, когда вы этого не хотите.

Я думаю, это не очень хорошее решение, если только кто-то не сможет его улучшить?

Я использую single_process на своем Gentoo;

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/1.0

Я столкнулся с этой проблемой на прошлой неделе, и, хотя я нашел несколько хороших решений, я решил сделать очень простой и чистый пакет python и загрузил его в PyPI. Он отличается от teno тем, что может заблокировать любое строковое имя ресурса. Хотя, конечно, можно заблокировать __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

Для всех, кто использует wxPython для своего приложения, вы можете использовать функцию wx.SingleInstanceCheckerзадокументировано здесь.

Я лично использую подкласс wx.App, который использует wx.SingleInstanceChecker и возвращает False из OnInit(), если существует существующий экземпляр приложения, уже выполняющийся следующим образом:

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

Вот мое возможное решение только для Windows. Поместите следующее в модуль, возможно, с именем onlyone.py или как-нибудь еще. Включите этот модуль непосредственно в ваш файл скрипта __ main __ 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)

Объяснение

Код пытается создать мьютекс с именем, производным от полного пути к сценарию. Мы используем косую черту, чтобы избежать путаницы с реальной файловой системой.

Преимущества

• Никакой конфигурации или «волшебных» идентификаторов не требуется, используйте их в любом количестве сценариев.
• Не осталось устаревших файлов, мьютекс умрет вместе с вами.
• Печатает полезное сообщение при ожидании

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

В системе Linux можно также спросить 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, если ограничение должно применяться к пользователям все. Отказ от ответственности: а) предполагается, что (базовое) имя скрипта уникально, б) могут быть условия гонки.

Лучшее решение для этого в Windows - использовать мьютексы, как предлагает @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 (включен также в пакет @sorin teno), который недоступен в Windows, и если вы попытаетесь заморозить свое приложение python с помощью такого пакета, как pyinstaller, который выполняет статический импорт, он выдает ошибку.

Кроме того, использование метода блокировки файла создает проблему read-only с файлами базы данных (это уже произошло с sqlite3).

Основываясь на ответе Роберто Росарио, я придумал следующую функцию:

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, поскольку она будет собираться только после завершения всего процесса. Если мы объявим в функции локальную переменную, она выйдет из области видимости после выхода из функции, и сокет будет удален.

Вся заслуга принадлежит Роберто Росарио, поскольку я только уточняю и уточняю его кодекс. И этот код будет работать только в Linux, как поясняет следующий цитируемый текст из https://troydhanson.github.io/network/Unix_domain_sockets.html:

Linux has a special feature: if the pathname for a UNIX domain socket begins with a null byte \0, its name is not mapped into the filesystem. Thus it won’t collide with other names in the filesystem. Also, when a server closes its UNIX domain listening socket in the abstract namespace, its file is deleted; with regular UNIX domain sockets, the file persists after the server closes it.

Поздний ответ, но для окон вы можете использовать:

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)

Вот кроссплатформенный пример, который я тестировал на Windows Server 2016 и Ubuntu 20.04 с использованием Python 3.7.9:

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:])

Вот хороший пример для django с contextmanager и memcached: https://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html

Может использоваться для защиты одновременной работы на разных хостах. Может использоваться для управления несколькими задачами. Также можно изменить для простых скриптов Python.

Моя модификация приведенного выше кода находится здесь:

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