레이블이 PIPE인 게시물을 표시합니다. 모든 게시물 표시
레이블이 PIPE인 게시물을 표시합니다. 모든 게시물 표시

python command timeout (Advanced)

python에서 command 실행시 타임아웃 설정 (Advanced)


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import subprocess as sp

class ExecuteFile:
    class TimeoutException(Exception):
        pass

    # timeout 커맨드실행
    @staticmethod
    def execute(cmd, errorKeyword=['error', 'fail', 'bad'], timeout=10):
        #pdb.set_trace()
        #popen = sp.Popen(cmd.split(), stderr=sp.STDOUT, stdout=sp.PIPE, env=os.environ)
        popen = sp.Popen(cmd, shell=True, stderr=sp.STDOUT, stdout=sp.PIPE, env=os.environ)

        # timeout 처리
        sTime = time.time()
        try:
            while True:
                if popen.poll() is not None:
                    break

                if time.time()-sTime > timeout:
                    raise ExecuteFile.TimeoutException

                time.sleep(0.1)

        except ExecuteFile.TimeoutException as e:
            popen.kill()

            return False, cmd, 'TimeoutException'

        # 에러 keyword 처리
        stdout = popen.stdout.read()

        for keyword in errorKeyword:
            if keyword in stdout:
                return False, cmd, stdout


        return True, cmd, 'SUCCESS'

python command timeout

python에서 command 실행시 타임아웃 설정

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import subprocess

'''
command 실행시 타임아웃을 걸고 stdout은 PIPE로 출력하여
'RESULT'가 검색된다면 'RESULT' 아래 라인을 반환
'''

def executeCmdTimeout( strCmd, timeout=60 ):
        # subprocess.Popen은 명령어는 실행시 fork하여 실행됨.
        p = subprocess.Popen( strCmd, stdout=subprocess.PIPE, shell=True )
        startTime = time.time()

        while p.poll() is None:
                time.sleep(0.1) # 100ms 마다 폴링하여 시간 확인

                nowTime = time.time()
                if nowTime - startTime > timeout: # 타임아웃이 되었다면
                        p.kill()
                        return ''

        lines = p.stdout.read().split('\n') # line단위로 스플릿

        # stdout에 RESULT라는 스트링이 존재한다면 바로 아래 라인을 반환
        idx = 0
        while idx < len(lines):
                if lines[idx][:len('RESULT')] == 'RESULT':
                        return lines[idx+1:]

                idx += 1

        return None


>>>> advanced 코드글 링크