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

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

Простое, кроссплатформенное решение, которое можно найти в другой вопрос от 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

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

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

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

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

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

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

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

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

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

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

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

Вот мое возможное решение только для 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)

Объяснение

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

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

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

Лучшее решение для этого в 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).

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

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

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

Любой, кто использует 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()

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

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

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

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

В Linux есть особенность: если путь к сокету домена UNIX начинается с нулевого байта \ 0, его имя не отображается в файловой системе. Таким образом, он не будет конфликтовать с другими именами в файловой системе. Кроме того, когда сервер закрывает свой прослушивающий сокет домена UNIX в абстрактном пространстве имен, его файл удаляется; с обычными сокетами домена UNIX файл сохраняется после его закрытия сервером.

Я использую 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/

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

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

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

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

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

Я столкнулся с этой проблемой на прошлой неделе, и, хотя я нашел несколько хороших решений, я решил сделать очень простой и чистый пакет 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

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

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

пример 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

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

Вот хороший пример для 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()