Есть ли у Pythonic способ запустить только один экземпляр программы?
Единственное разумное решение, которое я придумала, - это попытаться запустить его как сервер на каком-либо порту, а затем вторая программа, пытающаяся привязаться к тому же порту, завершится неудачей. Но это не очень хорошая идея, может быть, есть что-то более легкое, чем это?
(Примите во внимание, что иногда ожидается сбой программы, т.е. 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. Пожалуйста, ошибки в файлах здесь.
Вы можете установить тенденцию, используя один из следующих способов:
easy_install tendopip install tendoПростой, кросс-платформенный решение, найденный в еще один вопрос: zgoda:
import fcntl, sys
pid_file = 'program.pid'
fp = open(pid_file, 'w')
try:
fcntl.lockf(fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError:
# another instance is running
sys.exit(0)
Во многом похоже на предложение S.Lott, но с кодом.
Этот код специфичен для 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 может быть изменена, чтобы разрешить нескольким программам, которым требуется один экземпляр.
Я не знаю, достаточно ли это достаточно pythonic, но в мире Java прослушивание определенного порта является довольно широко используемым решением, так как оно работает на всех основных платформах и не имеет проблем с сбоями программ.
Другим преимуществом прослушивания порта является то, что вы можете отправить команду запущенному экземпляру. Например, когда пользователи запускают программу во второй раз, вы можете отправить исполняемому экземпляру команду, чтобы сообщить ей открыть другое окно (например, что делает Firefox). Я не знаю, используют ли они порты TCP или именованные каналы или что-то вроде этого ".)
Никогда не пиновал питон раньше, но это то, что я только что реализовал в 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) и поддерживает блокировку до тех пор, пока script не закончится (нормальный или ошибка).
Используйте файл 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())
Итак, другими словами, вы проверяете, существует ли pidfile; если нет, напишите pid в этот файл. Если pidfile существует, то проверьте, является ли pid pid запущенного процесса; если да, то у вас есть еще один живой процесс, поэтому просто выключите. Если нет, то предыдущий процесс разбился, поэтому запишите его, а затем напишите свой собственный pid в файл вместо старого. Затем продолжайте.
Вы уже нашли ответ на аналогичный вопрос в другом потоке, так что для полноты ознакомьтесь с тем, как добиться того же самого в Windows с помощью имени mutex.
Это может сработать.
Попытайтесь создать PID файл в известном месте. Если вы потерпите неудачу, у кого-то файл заблокирован, все готово.
Когда вы закончите нормально, закройте и удалите файл PID, чтобы кто-то другой мог его перезаписать.
Вы можете обернуть свою программу в оболочку script, которая удалит файл PID, даже если ваша программа выйдет из строя.
Вы также можете использовать PID файл, чтобы убить программу, если она зависает.
Использование lock файла - довольно распространенный подход к unix. Если он падает, вы должны очистить его вручную. Вы можете сохранить PID в файле, а при запуске проверить, есть ли процесс с этим PID, переопределяя файл блокировки, если нет. (Тем не менее, вам также нужна блокировка файла read-file-check-pid-rewrite). Вы найдете то, что вам нужно для получения и проверки pid в os -package. Общим способом проверки, существует ли процесс с данным pid, является отправка ему нефатального сигнала.
Другие альтернативы могли бы сочетать это с семафорами стада или 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()
Я отправляю это как ответ, потому что я новый пользователь, и Qaru не позволит мне проголосовать.
Решение 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), temp dir является /tmp для всех пользователей.
Это означает, что когда файл блокировки создается на многопользовательской машине, он создается в /tmp, и только пользователь, который создает файл блокировки, в первый раз сможет запустить приложение.
Возможное решение - вставить текущее имя пользователя в имя файла блокировки.
Стоит отметить, что решение OP для захвата порта также будет неверно работать на многопользовательской машине.
Я использую single_process на моем gentoo;
pip install single_process
пример:
from single_process import single_process
@single_process
def main():
print 1
if __name__ == "__main__":
main()
Вот мое возможное решение для Windows. Поместите следующее в модуль, возможно, называемый "onlyone.py" или что угодно. Включите этот модуль непосредственно в ваш __ основной файл __ python script.
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)
Код пытается создать мьютекс с именем, полученным из полного пути к script. Мы используем перекосы, чтобы избежать путаницы с реальной файловой системой.
Лучшее решение для этого на окнах - это использовать мьютексы, как предложено @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), который недоступен в Windows, и если вы попытаетесь заморозить свое приложение на python, используя пакет, такой как pyinstaller который выполняет статический импорт, он выдаст ошибку.
Кроме того, при использовании метода блокировки файла создается проблема, связанная с sqlite3 read-only для read-only файлам базы данных (в случае с sqlite3).
Я все время подозреваю, что это должно быть хорошим решением POSIXy, используя группы процессов, без необходимости попадания в файловую систему, но я не могу это сделать. Что-то вроде:
При запуске ваш процесс отправляет "kill -0" всем процессам в определенной группе. Если такие процессы существуют, он завершается. Затем он присоединяется к группе. Никакие другие процессы не используют эту группу.
Тем не менее, это условие гонки - многократные процессы могут сделать это в одно и то же время, и все в конечном итоге присоединяются к группе и работают одновременно. К тому времени, когда вы добавили какой-то мьютекс, чтобы сделать его водонепроницаемым, вам больше не нужны группы процессов.
Это может быть приемлемым, если ваш процесс запускается только cron один раз в минуту или каждый час, но он меня немного нервничает, что он ошибается точно в тот день, когда вы этого не хотите.
Я думаю, это не очень хорошее решение в конце концов, если кто-то не может улучшить его?
Я столкнулся с этой точной проблемой на прошлой неделе, и хотя я нашел несколько хороших решений, я решил сделать очень простой и чистый пакет python и загрузить его в PyPI. Он отличается от tendo тем, что он может блокировать любое имя строкового ресурса. Хотя вы могли бы заблокировать __file__ для достижения того же эффекта.
Установить с помощью: pip install quicklock
Использование очень просто:
[[email protected] ~/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 create a lock so that only one instance of a script will run
...
>>> singleton('hello world')
>>>
>>> # Let 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
[[email protected] ~/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
В системе 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 если ограничение должно распространяться на всех пользователей. Отказ от ответственности: а) предполагается, что имя сценария (базовое) является уникальным, б) могут быть условия гонки.
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()
Я бы просто сделал что-то вроде:
if commands.getstatusoutput("mkdir /tmp/test")[0]:
print "Exiting"
sys.exit(1)
И где-то в конце моего кода я удалю каталог. mkdir является атомарным и работает здесь очень хорошо.