Verifique se apenas uma única instância de um programa está sendo executada

120

Existe uma maneira Pythonic de ter apenas uma instância de um programa em execução?

A única solução razoável que eu encontrei é tentar executá-lo como um servidor em alguma porta e, em seguida, o segundo programa tentar ligar à mesma porta - falha. Mas não é realmente uma ótima idéia, talvez haja algo mais leve que isso?

(Leve em consideração que, às vezes, espera-se que o programa falhe, ou seja, segfault - para que coisas como "bloquear arquivo" não funcionem)

Slava V
fonte
1
Talvez sua vida fosse mais fácil se você localizasse e corrigisse o segfault. Não que seja uma coisa fácil de fazer.
David Locke
Não está na minha biblioteca, está nas ligações libxml do python e é extremamente tímido - é acionado apenas uma vez a cada dois dias.
Slava V
5
A biblioteca padrão do Python suporta flock (), que é a coisa certa para programas UNIX modernos. A abertura de uma porta usa um ponto em um espaço para nome muito mais restrito, enquanto os arquivos pid são mais complexos, pois você precisa verificar os processos em execução para invalidá-los com segurança; rebanho não tem nenhum problema.
Charles Duffy
s / UNIX / linux / pronto, FTFY.
precisa saber é o seguinte

Respostas:

100

O código a seguir deve fazer o trabalho, é multiplataforma e é executado no Python 2.4-3.2. Eu testei no Windows, OS X e Linux.

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

A versão mais recente do código está disponível singleton.py . Por favor, registre erros aqui .

Você pode instalar tend usando um dos seguintes métodos:

sorin
fonte
2
Atualizei a resposta e incluí um link para a versão mais recente. Se você encontrar um bug, envie-o para o github e eu o solucionarei o mais rápido possível.
sorin
2
@Johny_M Graças, eu fiz um patch e lançou uma versão mais recente no pypi.python.org/pypi/tendo
Sorin
2
Essa sintaxe não funcionou para mim no Windows no Python 2.6. O que funcionou para mim foi: 1: de tendo import singleton 2: me = singleton.SingleInstance ()
Brian
25
Outra dependência de algo tão trivial quanto isso? Não parece muito atraente.
precisa saber é o seguinte
2
O singleton lida com processos que recebem um sigterm (por exemplo, se um processo está sendo executado por muito tempo), ou eu tenho que lidar com isso?
JimJty
43

Solução simples, multiplataforma , encontrada em outra pergunta pelo 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

Muito parecido com a sugestão de S.Lott, mas com o código.

Slava V
fonte
Por curiosidade: isso é realmente multiplataforma? Isso funciona no Windows?
Joachim Sauer
1
Não há fcntlmódulo no Windows (embora a funcionalidade possa ser emulada).
JFS
10
DICA: se você deseja agrupar isso em uma função, 'fp' deve ser global ou o arquivo será fechado após a saída da função.
Cmcginty
1
O @Mirko Control + Z não sai de um aplicativo (em qualquer sistema operacional que eu conheça), ele o suspende. O aplicativo pode ser retornado ao primeiro plano com fg. Portanto, parece que está funcionando corretamente para você (ou seja, o aplicativo ainda está ativo, mas suspenso, para que o bloqueio permaneça no lugar).
Sam Bull
1
Este código na minha situação (Python 3.8.3 no Linux) precisava de modificação:lock_file_pointer = os.open(lock_path, os.O_WRONLY | os.O_CREAT)
baziorek 16/06
30

Este código é específico do Linux. Ele usa soquetes de domínio UNIX 'abstratos', mas é simples e não deixa arquivos de bloqueio antigos. Eu prefiro a solução acima porque não requer uma porta TCP especialmente reservada.

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) 

A cadeia exclusiva postconnect_gateway_notify_lockpode ser alterada para permitir vários programas que precisam de uma única instância imposta.

Roberto Rosario
fonte
1
Roberto, você tem certeza de que após o pânico do kernel ou a reinicialização total, o arquivo \ 0postconnect_gateway_notify_lock não estará presente na inicialização? No meu caso, o arquivo de soquete AF_UNIX ainda está presente depois disso e isso destrói toda a idéia. A solução acima com a aquisição de bloqueio em um nome de arquivo específico é muito confiável nesse caso.
Daniel Gurianov
2
Como mencionado acima, esta solução funciona em Linux, mas não no Mac OS X.
Bilal e Olga
2
Esta solução não funciona. Eu tentei no Ubuntu 14.04. Execute o mesmo script a partir de 2 janelas de terminal simultaneamente. Ambos correm muito bem.
Dimon
1
Isso funcionou para mim no Ubuntu 16. E matar o processo de qualquer maneira permitiu que outro iniciasse. Dimon, acho que você fez algo errado no seu teste. (Talvez você tenha esquecido de fazer seu script dormir após a execução do código acima, por isso, imediatamente saiu e liberou o soquete.) #
470 Luke Luke
1
Não é uma questão de dormir. O código funciona, mas apenas como código embutido. Eu estava colocando isso em uma função. O soquete estava desaparecendo assim que a função existia.
Steve Cohen
25

Não sei se é python o suficiente, mas no mundo Java, ouvir em uma porta definida é uma solução bastante usada, pois funciona em todas as principais plataformas e não tem problemas com programas que causam pane.

Outra vantagem de ouvir uma porta é que você pode enviar um comando para a instância em execução. Por exemplo, quando os usuários iniciam o programa pela segunda vez, você pode enviar à instância em execução um comando para dizer a ele para abrir outra janela (é o que o Firefox faz, por exemplo. Não sei se eles usam portas TCP ou pipes nomeados ou algo assim ').

Joachim Sauer
fonte
+1 a isso, especialmente porque ele permite que eu notifique a instância em execução, para que ela crie outra janela, apareça etc.
noto Hugo
1
Use, por exemplo import socket; s = socket.socket(socket.AF_INET, socket.SOCK_STREAM); s.bind(('localhost', DEFINED_PORT)). Um OSErrorserá gerado se outro processo estiver vinculado à mesma porta.
crishoj
12

Nunca escrevi python antes, mas é isso que acabei de implementar no mycheckpoint, para impedir que ele seja iniciado duas ou mais vezes por 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()

Encontrei a sugestão do Slava-N depois de postar isso em outra edição (http://stackoverflow.com/questions/2959474). Este é chamado como uma função, bloqueia o arquivo de scripts em execução (não um arquivo pid) e mantém o bloqueio até o final do script (normal ou erro).

MD Klapwijk
fonte
Muito elegante. Eu mudei para que ele obtenha o caminho dos argumentos do script. Também recomenda incorporar isso em algum lugar comum - Exemplo
Jossef Harush
10

Use um arquivo pid. Você tem um local conhecido, "/ path / to / pidfile" e, na inicialização, faz algo assim (parcialmente pseudocódigo porque eu sou pré-café e não quero trabalhar tanto):

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

Portanto, em outras palavras, você está verificando se existe um arquivo pid; caso contrário, escreva seu pid nesse arquivo. Se o arquivo pid existir, verifique se o pid é o pid de um processo em execução; Nesse caso, você tem outro processo ativo em execução, então desligue-o. Caso contrário, o processo anterior falhou, registre-o e, em seguida, escreva seu próprio pid no arquivo no lugar do antigo. Então continue.

Charlie Martin
fonte
4
Isso tem uma condição de corrida. A sequência de teste e gravação pode gerar uma exceção de dois programas iniciados quase simultaneamente, não encontrar nenhum arquivo e tentar abrir para gravação simultaneamente. Isso deve gerar uma exceção em um, permitindo que o outro prossiga.
21468 S.Lott
5

Isso pode funcionar.

  1. Tente criar um arquivo PID para um local conhecido. Se você falhar, alguém tiver o arquivo bloqueado, está pronto.

  2. Quando terminar normalmente, feche e remova o arquivo PID, para que outra pessoa possa substituí-lo.

Você pode agrupar seu programa em um script de shell que remove o arquivo PID, mesmo se o seu programa travar.

Você também pode usar o arquivo PID para matar o programa se ele travar.

S.Lott
fonte
3

Usar um arquivo de bloqueio é uma abordagem bastante comum no unix. Se travar, você precisará limpar manualmente. Você pode armazenar o PID no arquivo e, na inicialização, verificar se existe um processo com esse PID, substituindo o arquivo de bloqueio, se não houver. (No entanto, você também precisa de um bloqueio em torno do arquivo read-file-check-pid-rewrite-file). Você encontrará o que precisa para obter e verificar o pid no sistema operacional pacote . A maneira comum de verificar se existe um processo com um determinado pid é enviar um sinal não fatal.

Outras alternativas poderiam ser combinadas com semáforos flock ou posix.

Abrir um soquete de rede, como a saua propôs, provavelmente seria o mais fácil e mais portátil.

Rolf Rander
fonte
3

Para quem usa o wxPython para seu aplicativo, você pode usar a função wx.SingleInstanceChecker documentada aqui .

Eu pessoalmente uso uma subclasse de wx.Appque faz uso de wx.SingleInstanceCheckere retornos Falsede OnInit()se existe uma instância existente do aplicativo já execução assim:

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

Este é um substituto simples wx.Appque proíbe várias instâncias. Para usá-lo, basta substituir wx.Apppor SingleAppno seu código da seguinte forma:

app = SingleApp(redirect=False)
frame = wx.Frame(None, wx.ID_ANY, "Hello World")
frame.Show(True)
app.MainLoop()
Matt Coubrough
fonte
Depois de codificar um encadeamento de soquete para um singleton, achei isso, que funciona muito bem e eu já instalei em um programa, no entanto, eu gostaria da "ativação" adicional que posso dar ao singleton para que eu possa levá-lo ao diretório frente e centro de uma grande pilha de janelas sobrepostas. Também: o "documentado aqui" link aponta documentação gerada automaticamente bastante inútil esta é uma melhor ligação
RufusVS
@RufusVS Você está certo - esse é um link de documentação muito melhor, atualizei a resposta.
precisa saber é o seguinte
3

Aqui está minha eventual solução apenas para Windows. Coloque o seguinte em um módulo, talvez chamado de 'onlyone.py' ou qualquer outra coisa. Inclua esse módulo diretamente no seu arquivo de script python __ main __.

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)

Explicação

O código tenta criar um mutex com o nome derivado do caminho completo para o script. Usamos barras para evitar possíveis confusões com o sistema de arquivos real.

Vantagens

  • Nenhuma configuração ou identificadores 'mágicos' são necessários, use-o em quantos scripts diferentes forem necessários.
  • Sem arquivos antigos, o mutex morre com você.
  • Imprime uma mensagem útil ao aguardar
Keeely
fonte
3

A melhor solução para isso no Windows é usar mutexes, conforme sugerido por @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")

Algumas respostas usam fctnl (incluído também no pacote @sorin tendo) que não está disponível no Windows e, se você tentar congelar seu aplicativo python usando um pacote como o pyinstallerque faz importações estáticas, gera um erro.

Além disso, o uso do método de bloqueio de arquivo cria um read-onlyproblema com os arquivos de banco de dados (com experiência nisso sqlite3).

Chuck G
fonte
2

Estou postando isso como resposta porque sou um novo usuário e o Stack Overflow não me permite votar ainda.

A solução de Sorin Sbarnea funciona para mim no OS X, Linux e Windows, e sou grato por isso.

No entanto, tempfile.gettempdir () se comporta de uma maneira no OS X e Windows e outro sob outros / muitos / todos (?) * Nixes (ignorando o fato de o OS X também ser o Unix!). A diferença é importante para esse código.

O OS X e o Windows têm diretórios temporários específicos do usuário, portanto, um arquivo temporário criado por um usuário não fica visível para outro usuário. Por outro lado, em muitas versões do * nix (testei o Ubuntu 9, RHEL 5, OpenSolaris 2008 e FreeBSD 8), o diretório dir é / tmp para todos os usuários.

Isso significa que, quando o arquivo de bloqueio é criado em uma máquina multiusuário, ele é criado em / tmp e somente o usuário que cria o arquivo de bloqueio pela primeira vez poderá executar o aplicativo.

Uma solução possível é incorporar o nome de usuário atual no nome do arquivo de bloqueio.

Vale ressaltar que a solução do OP de pegar uma porta também se comportará mal em uma máquina multiusuário.

Philip Semanchuk
fonte
Para alguns leitores (por exemplo, eu), o comportamento desejado é que apenas uma cópia possa executar um período, independentemente de quantos usuários estejam envolvidos. Portanto, os diretórios tmp por usuário são quebrados, enquanto o / tmp compartilhado ou o bloqueio de porta exibem o comportamento desejado.
27613 Jonathan Hartley
2

Eu uso single_processno meu gentoo;

pip install single_process

exemplo :

from single_process import single_process

@single_process
def main():
    print 1

if __name__ == "__main__":
    main()   

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

gkiwi
fonte
Falha no Py3. O pacote parece mal interpretado.
Ekevoo
No Windows eu recebo: ImportError: No módulo chamado fcntl
Andrew W. Phillips
1

Eu continuo suspeitando que deveria haver uma boa solução POSIXy usando grupos de processos, sem ter que acessar o sistema de arquivos, mas não consigo identificá-lo. Algo como:

Na inicialização, seu processo envia um 'kill -0' para todos os processos em um grupo específico. Se existir algum desses processos, ele será encerrado. Então ele se junta ao grupo. Nenhum outro processo usa esse grupo.

No entanto, isso tem uma condição de corrida - vários processos podem fazer isso exatamente ao mesmo tempo e acabam ingressando no grupo e funcionando simultaneamente. No momento em que você adicionou algum tipo de mutex para torná-lo à prova d'água, você não precisa mais dos grupos de processos.

Isso pode ser aceitável se o seu processo for iniciado apenas pelo cron, uma vez a cada minuto ou a cada hora, mas me deixa um pouco nervoso que isso daria errado exatamente no dia em que você não deseja.

Acho que essa não é uma solução muito boa, a menos que alguém possa melhorar isso?

Jonathan Hartley
fonte
1

Encontrei exatamente esse problema na semana passada e, embora tenha encontrado boas soluções, decidi criar um pacote python muito simples e limpo e o enviei para o PyPI. Difere de ter, pois pode bloquear qualquer nome de recurso de cadeia. Embora você certamente possa bloquear__file__ para obter o mesmo efeito.

Instale com: pip install quicklock

Usá-lo é extremamente simples:

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

Dê uma olhada: https://pypi.python.org/pypi/quicklock

Nate Ferrero
fonte
1
Acabei de instalar via "pip install quicklock", mas quando tento usá-lo via "from quicklock import singleton", recebo uma exceção: "ImportError: não é possível importar o nome 'singleton'". Isso está em um Mac.
grayaii
Acontece que o quicklock não funciona com o python 3. Essa é a razão pela qual estava falhando para mim.
grayaii
Sim, desculpe, não foi à prova de futuro. Congratularei-me com uma contribuição para fazê-lo funcionar!
Nate Ferrero
1

Com base na resposta de Roberto Rosario, proponho a seguinte função:

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

Precisamos definir a SOCKETdisponibilidade global, pois ela só será coletada como lixo quando todo o processo for encerrado. Se declararmos uma variável local na função, ela ficará fora do escopo após a saída da função, portanto, o soquete será excluído.

Todo o crédito deve ser para Roberto Rosario, já que apenas esclareço e elaborei seu código. E esse código funcionará apenas no Linux, como o seguinte texto citado em https://troydhanson.github.io/network/Unix_domain_sockets.html explica:

O Linux possui um recurso especial: se o nome do caminho para um soquete de domínio UNIX começar com um byte nulo \ 0, seu nome não será mapeado no sistema de arquivos. Portanto, ele não colidirá com outros nomes no sistema de arquivos. Além disso, quando um servidor fecha seu soquete de escuta de domínio UNIX no espaço para nome abstrato, seu arquivo é excluído; com soquetes de domínio UNIX regulares, o arquivo persiste após o servidor o fechar.

makiko_fly
fonte
0

exemplo linux

Este método é baseado na criação de um arquivo temporário excluído automaticamente após o fechamento do aplicativo. no lançamento do programa, verificamos a existência do arquivo; se o arquivo existe (há uma execução pendente), o programa é fechado; caso contrário, ele cria o arquivo e continua a execução do programa.

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
Kerwal
fonte
1
Bem-vindo ao Stack Overflow! Embora essa resposta possa estar correta, adicione algumas explicações. Impartir a lógica subjacente é mais importante do que apenas fornecer o código, porque ajuda o OP e outros leitores a resolver esses problemas e outros similares.
CodeMouse92
Este thread é seguro? Parece que a verificação ea criação arquivo temporário não são atômica ...
coppit
0

Em um sistema Linux, também é possível solicitar pgrep -ao número de instâncias, o script é encontrado na lista de processos (a opção -a revela a cadeia de caracteres completa da linha de comandos). Por exemplo

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

Remova -u $UIDse a restrição se aplicar a todos os usuários. Isenção de responsabilidade: a) presume-se que o nome do script (base) seja único; b) pode haver condições de corrida.

user71769
fonte
-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()  
Ertan Özer
fonte
2
Bem-vindo ao Stack Overflow! Embora esse bloco de código possa responder à pergunta, seria melhor se você pudesse fornecer uma pequena explicação sobre o motivo. Por favor edite sua resposta para incluir tal descrição.
Artjom B.