calInThread vs callFromThread

twistedmatrix에서 callInThread()와 callFromThread()는 어떻게 다른 것일까?
무심코, 코드에서 callInThread()를 썼는데, 그 차이를 궁금해 하다가 문서 및 소스를 찾아보게
되었다.

1. The Twisted Documentation 보기
우선 살펴본, The Twisted Documentation에서는 callFromThread()는 thread-safe하지 않은 메소드를 thread-safe하게 실행시킬 때 사용한다고 한다. 다른 설명으로는, thread-safe하지 않은 메소드를 메인 이벤트 루프에서 실행되도록 스케쥴한다고 한다.

from twisted.internet import reactor

def notThreadSafe(x):

   “”” do something that isn’t thread-safe “””

   # …
def threadSafeScheduler():

   “”” Run in thread-safe manner.”””

   reactor.callFromThread(noThreadSafe, 3)

반면, callInThread()는 thread상에서 메소드들을 호출하고 싶을 때 사용하며, 이 메소드들이 thread pool에 의해 queue처리 되면서 실행되도록 한다고 한다.

from twisted.internet import reactor

def aSillyBlockingMethod(x):
   import time
   time.sleep(2)
   print x

# run method in thread
reactor.callInThread(aSillyBlockiingMethod, “2 seconds have passed”)

이것을 본 소감은..

1. “그게 뭔데~ 어쩌라고~”
2. no thread-safe 함수는 callFromThread()를 쓰고, safe 함수는 callInThread()를 쓰는 것인가?
3. blocking이 내부적으로 발생할 수 있는 함수는 callInThread()를 쓰는 것인가?

등 용도가 명확하지 않았다.

---

2. API 찾아보기
그래서, twisted API로부터, callInThread()와 callFromThread()에 대한 설명을 참고하게 되었다.
아래는 각각 callFromThread()와 callInThread()에 대한 API 설명이다.

[callFromThread()]

Run the callable object in a separate thread.

[callInThread()]

Cause a function to be executed by the reactor thread.

Use this method when you want to run a function in the reactor’s thread from another thread. Calling callFromThread should wake up the main thread (where reactor.run() is executing) and run the given callable in that thread.

Obviously, the callable must be thread safe. (If you want to call a function in the next mainloop iteration, but you’re in the same thread, use callLater with a delay of 0.)

위의 설명을 따른다면, callFromThread()는 별도의 thread를 띄워서 실행시켜 준다는 것이나,
callInThread()는 reactor의 내부 thread에 위탁한다는 것으로 보인다.

---

3. 다른 참고 문서에서는..
한편, 다른 문서에서는 각 메소드의 사용 용도에 차이를 다음과 같이 두었다.

def threadDoSomething(a, b):
   d = defer.Deferred()
   def func():
      result = doSomething(a, b)
      reactor.callFromThread(d.callback, result)
   t = threading.Thread(target = func)
   t.start()
   return d

def threadDoSomethingNew(a, b):
   d = defer.Deferred()
   def func():
      result = doSomething(a, b)
      reactor.callFromThread(d.callback, result)
   reactor.callInThread(func)
   return d

위의 사용예와 해당 문서의 설명에 따르면, reactor.callFromThread()는 twisted 중, 유일하게 thread-safe 메소드이며, reactor thread의 내부에서 주어진 함수를 실행시킨다. 첫번째 함수는 정상동작하나, 반복적으로 호출하면 새로운 thread를 계속 생성하므로 시스템에 부하를 주므로, 두번째 함수와 같이 사용하는 것이 효과적이다.

결과적으로, callInThread()는  twisted를 사용하지 않는 방식에서 threading.Thread().start() 대신에 사용하면 된다는 것이다.

from twisted.internet import reactor

c = 1
# this triial example function is
# not thread-safe
def increment(x):
   global c
   c = c + x

def threadTask():
   # run non-threadsafe function
   # in thread-safe way — increment(7)
   # will be called in event loop thread
   reactor.callFromThread(increment, 7)

# add a number of tasks to thread pool
for i in range(10):
   reactor.callInThread(threadTask)

출처 : Network Programming for the Rest of Us, Glyph Lefkowitz, http://www.twistedmatrix.com/users/glyph

위의 예 대로라면, thread를 launch 시킬 때는 callInThread()를 쓰고, callFromThread()를 사용하여 thread-safe하게 동작하도록 해준다는 것이다.

그래도, 미심쩍은 면은 있다!!

4. 소스에서는?
한편, 다른 문서에서는 각 메소드의 사용 용도에 차이를 다음과 같이 두었다.

[callFromThread]

추적해보니, base.py에 정의되어 있음.

def callFromThread(self, f, *args, **kw):
    “””See twisted.internet.interfaces.IReactorThreads.callFromThread.
    “””
    assert callable(f), “%s is not callable” % (f,)
    # lists are thread-safe in CPython, but not in Jython
    # this is probably a bug in Jython, but until fixed this code
    # won’t work in Jython.
    self.threadCallQueue.append((f, args, kw))
    self.wakeUp()

threadCallQueue는 ReactorBse에 선언된 변수로서, list이다. callFromThread()가 호출될 때마다
내부의 threadCallQueue 리스트에 추가되고, runUntilCurrent()라는 내부 메소드가 실행 되면 wakeup 해서, queue 내부의 함수를 순차적으로 시행하고, dequeue시킨다.

[callInThread]
추적해보니, base.py 및 threadpool.py에 정의되어 있음.

def callInThread(self, _callable, *args, **kwargs):
    “””See twisted.internet.interfaces.IReactorThreads.callInThread.
    “””
    if self.threadpool is None:
    self._initThreadPool()
    self.threadpool.callInThread(_callable, *args, **kwargs)

def callInThread(self, func, *args, **kw):
    if self.joined:
    return
    ctx = context.theContextTracker.currentContext().contexts[-1]
    o = (ctx, func, args, kw)
    self.q.put(o)
    if self.started:
        self._startSomeWorkers()

위의 코드에서 보는 바와 같이, callInThread를 호출하면, 내부에 정의한 Queue.Queue 객체에
함수를 넣고, thread pool에 정의된 수에 따라 동작시킨다. 내부적으로 threading.Thread()를
미리 실행시켜서 queue로 보관하고 있다.

결과적으로, callFromThread를 사용하면, 내부의 이벤트 루프에 넣어서 실행되도록 하고, callInThread를 사용하면, 내부에 생성해 놓은 thread pool로부터 실행하는 것이다라는 설명이 맞다는 뜻 되겠다~

따라서, 새로 시작하는 thread를 callInThread에 넣어서 시작시키고, 내부적으로 callFromThread를 타도록 시키는 게 맞겠다. 문서에 나와 있는 바와 같이, twisted API는 callFromThread를 통해서 호출하라고 한 것도 같은 맥락으로 보인다.

현재 클래스 이름, 메소드 이름, 라인 넘버 얻기

현재 실행하는 클래스의 이름과 메소드 이름, 라인 넘버를 얻어보자.

클래스의 인스턴스에서 호출한다는 가정하에,

def trace(obj, toList = False):
    import sys
    className = obj.__class__.__name__
    methodName = callersname()
    lineNumber = sys.exc_info()[2].tb_lineno
   
    if toList:
        return className, methodName, lineNumber
    else:
        return “%s.%s[%d]” % (className, methodName, lineNumber)

단, callersname()은

def callersname():
   return sys._getframe(2).f_code.co_name

그러나, 호출하는 클래스가 특정 클래스를 상속한 경우,
부모에 정의된 함수를 자식 클래스에서 그대로 사용하면서 trace()를 호출하였다면,
부모에 정의된 메소드를 실행하고 있더라도, 자식 클래스의 이름을
현재 클래스의 이름으로 리턴한다.

즉, A에 a()가 정의되어 있고, b = B(A)이나, b.a()를 실행도중 trace()를 호출하면,
현재 클래스의 이름이 A로 출력되는 대신 B로 출력된다.

현재 함수의 이름 얻기

# 현재 함수의 이름 얻기
def whoami():
    import sys
    return sys._getframe(1).f_code.co_name

# 현재 함수의 caller 이름 얻기
def callersname():
    import sys
    return sys._getframe(2).f_code.co_name

출처 : Python Recipe 66062: Determining Current Function Name

How to catch ‘ExpatError exception’ (handling)

ElementTree와 같은 패키지를 사용하여 XML를 파싱하는 경우,
XML 엘리먼트의 짝이 안맞는 등, 유효하지 않은 XML 구성이 탐지되면
ExpatError가 뜨는데,

try:
  # XML 연산
catch ExpatError, e:
  # do something

하면,
NameError: global name ‘ExpatError’ is not defined라는 에러가 뜬다.

이를 해결하려면,
ExpatError를 catch하는 py 파일의 앞에, scope을 맞춰서

from xml.parsers.expat  import ExpatError

라고 넣어주자. 그러면, 문제 해결!!

”
While you are using XML packages such as ElementTree, you might want to handle
ExpatError which might be caused by invalid-format or missing tags and so on.

However, you will see another error message saying “NameError: global name ‘ExpatError’
is not defined”.  What you only have to solve this problem is just add a line on the top of
your code.

That is,

from xml.parsers.expat import ExpatError

It’ll work, then.

queue의 push/pop 속도 비교

ㅁ Queue.Queue
   1. push(‘a’) : 6.39s
   2. push(‘a’) + pop() : 11.37s

ㅁ collections.dequeue
   1. append(‘a’) : 0.13s
   2. append(‘a’) + popleft() : 0.27s

ㅁ list
  1. append(‘a’) : 0.15s
  2. append(‘a’) + pop() : 0.40s

Queue를 쓰려면, dequeue를 활용하는 것이 좋을 것 같다.

python class의 메소드별 단위 테스트 (unittest pyUnit vs py.test)

여러가지 단위 테스트 프로그램이 존재하나,
그 중, standard library로 포함되어 있는 pyUnit (unittest) 과 py.test를 비교하여 간략하게
비교한다.

ㅁ 테스트를 위한 클래스

class A:
    def getA(self):
        return ‘a’

    def getWrongA(self):
        return ‘b’

if __name__ == ‘__main__’:
    aa = A()
    print aa.getA()

* 위의 코드에서 보는 바와 같이, 테스트 메소드는 두 개(getA와 getWrongA)이다.
* 각 메소드는 정상 상황(‘a’를 리턴할 것이라 예상하는 상황에서 실제로 ‘a’를 리턴)과 오류 상황(‘a’를 리턴할 것이라 예상하는 상황에서 실제로는 ‘a’가 아닌 다른 값을 리턴)을 나타낸다.

ㅁ 설치 방법
  1. pyUnit : 별도의 설치 과정 불필요
  2. py.test : 별도의 설치 과정 필요 (http://pypi.python.org/pypi/py 참고)

ㅁ 테스트 코드 작성 방법
   1. pyUnit

from a import A

import unittest

class ATester(unittest.TestCase):
    def setUp(self):
        print ‘setUp called..’
        self.aa = A()

    def testGetA(self):
        print ‘testGetA..’
        self.assertEquals(self.aa.getA(), ‘a’)

    def testGetWrongA(self):
        print ‘testGetWrongA..’
        self.assertEquals(self.aa.getWrongA(), ‘a’)

    def tearDown(self):
        print ‘tearDown called..’

if __name__ == ‘__main__’:
    testSuite = unittest.TestSuite()
    for testmethod in (‘testGetA’, ‘testGetWrongA’):
        testSuite.addTest(ATester(testmethod))

    runner = unittest.TextTestRunner()
    runner.run(testSuite)

   2. py.test

from a import A

def test_getA():
    aa = A()
    assert aa.getA() == ‘a’

def test_getWrongA():
    aa = A()
    assert aa.getWrongA() == ‘b’

ㅁ 실행 방법
  1. pyUnit

python test_with_unittest.py  (위의 예제 파일 이름이 test_with_unittest.py임)

  2. py.test

py.test (이름 지정하지 않아도 찾아서 테스트 실시)

ㅁ 실행 결과
1. pyUnit

setUp called..
testGetA..
tearDown called..
.setUp called..
testGetWrongA..
FtearDown called..

======================================================================
FAIL: testGetWrongA (__main__.ATester)
———————————————————————-
Traceback (most recent call last):
  File “test_with_unittest.py”, line 16, in testGetWrongA
    self.assertEquals(self.aa.getWrongA(), ‘a’)
AssertionError: ‘b’ != ‘a’

———————————————————————-
Ran 2 tests in 0.000s

FAILED (failures=1)

2. py.test

================================ test process starts ================================
executable:   /usr/bin/python  (2.5.2-final-0)
using py lib: /root/jhkim/lib/py-0.9.2/py <rev unknown>

test_with_py.py[2] .F

======================================================================================
________________________________entrypoint: test_getWrongA _____________________________
    def test_getWrongA():
        aa = A()
E       assert aa.getWrongA() == ‘a’
>       assert ‘b’ == ‘a’
         +  where ‘b’ = <a.A instance at 0x838392c>.getWrongA()

/root/test/test_with_py.py:9: AssertionError
_______________________________________________________________________________________
tests finished: 1 passed, 1 failed in 0.18 seconds ========================================================

 사용 목적이나 규모에 따라, 사용자 기호에 따라 각각 장단점이 존재한다!

현재 클래스의 이름 얻기 + 현재 클래스의 메소드 얻기

1. 현재 클래스의 메소드 이름 얻기

class Base 에 대해,

Base.__dict__ 라고 하면, Base에 선언된 정보를 얻을 수 있으나,
이중, 메소드 이름만 추출하고 싶다면,

from types import *

def getMethodNames():
  result = []
  for attr, val in Base.__dict__.items():
    if type(val) == FunctionType:
       result.append(attr)
  return result

과 같이 함으로써, method 이름을 추출할 수 있다.

2. 그러나,  이미 특정 class로부터 instance를 만든 경우에는 위의 방법을 그대로 사용할 수 없다.
즉, Base.getMethodNames()라고 하면 결과를 얻을 수 있지만,

aa = Base()
aa.getMethodNames()

라고 하면, 데이터를 얻을 수 없다.

따라서, aa 로부터 instance하기 전의 클래스가 무엇인지 알아내야 한다.

self.__class__

라고 하면, 현재의 클래스를 리턴한다.

따라서, 위의 for 문에서

Base.__dict__items()  대신, self.__class__.__dict__.items()

라고 쓰면 된다.

이때, 클래스의 이름을 self.__class__.__name__ 으로 하여 확인할 수도 있는데,

self.__class__.__dict__.itesm() 와 eval(self.__class__.__name__).__dict__.items()는
동일한 결과물을 뽑아준다.

즉, class와 eval(class의 이름)이 동일하게 취급된다는 것.

[Twisted] adpapi – Query 실행하기/ CallBack 함수에 argument 전달하기

일단, Query를 실행하는 방법은 크게 두 가지를 쓸 수 있겠다.

runQuery()를 사용하는 방법과, runInteraction()을 사용하는 방법이다.

주어진 table에서 10개의 row를 가져오고자 한다고 했을때,

1) runQuery() 사용하기

def getCount():

return dbpool.runQuery(“select * from test”)

def printResult(l):

if l:

result = l[0][0]
print l[0][0], ” records”

else:

print “no rows fetched”

dbpool = adbapi.ConnectionPool(‘MySQLdb’, db=’abcd’, user=’abcd’, passwd=’abcd’)
getCount().addCallback(printResult)
reactor.run()

2) runInteraction() 사용하기

def get10():

return dbpool.runInteraction(_get10)

def _get10(txn):

txn.execute(“select * from test”)
result = txn.fetchall()

if result:

return result

else:

return None

def print10(data):

if data != None:

print data, ‘ data found’

else:

print ‘no data’

 

dbpool = adbapi.ConnectionPool(‘MySQLdb’, db=’abcd’, user=’abcd’, passwd=’abcd’)
getCount().addCallback(printResult)

get10().addCallback(print10)
reactor.run()

위와 같은 서로 다른 방법으로 구성할 수 있다. 이때, runInteraction을 사용하면, function에 대해 실행할 수 있고, transaction 단위로 실행시킬 수 있다는 점이다. 

또한, 별도의 메타정보가 함께 결과에 포함되지 않는 것 같으므로, 메타 정보를 query를 실행할 때 미리 주고 싶다면??
 
이렇게 하는 수도 있겠다. 실행이 synchronous 하지 않으므로, 어떤 필드가 어떻게 매핑되어야 할지는 query와 함께 움직여야 하지 않을까? 그래서, Callback 함수에 테이블의 메타 정보를 한번 심어봤다. runQuery()와, runInteraction()에 각각. 동작이 하는 경우는 runInteraction()의 경우이다.

예를 들어, 넘기고자 하는 파라미터가 list l = (‘a’,’b’,’c’,’d’)라고 하면, 각각의 함수는 다음과 같이 바뀔 것이다.

1. runQuery()의 경우

def getCount(param-list):

return dbpool.runQuery(“select * from test”, param_list)

def printResult(l, param_list):

# 동일

그러나, 이 경우는 에러가 발생한다. runQuery의 파라미터로 쿼리문과 함께 사용자 정의 파라미터를 전달하고 싶지만, printResult callback 문으로 전달될 수 없다. (물론 방법이 있겠지만, 나는 모르겠다. 아직은 ^^;)

한편, 이 상황을 runInteraction()을 사용하는 경우에는 해결 가능한 문제가 된다.

2. runInteraction()의 경우

def get10(param_list):

return dbpool.runInteraction(_get10, param_list)   # _get10 파라미터로 param_list 지정

def _get10(txn, param_list):

# param_list로 하고 싶은 작업을 할 수 있다
txn.execute(“select * from test”)
result = txn.fetchall()
if result:

return [param_list, result]

else:

return [param_list, None]

# 결과를 list로 return 한다.

def print10(data):

if data != None:

meta = data[0]      # param_list from _get10
record = data[1]    # result from _get10

get10()이나 _get10()에 다른 형태로 param_lsit를 곧바로 넘기려고 하면, tuple에 addCallback할
수 없다는 에러가 뜨거나, 원하는 형태로 전달할 수 없음을 확인했다.

따라서, 위의 방법과 같이, result와 넘겨주려는 데이터를 함께 리스트로 넘겨서 직접 확인하는
방법을 구상해보았다. (리스트 또는 dict 형태로 하면 좋을 것 같다.) 더 좋은 방법이 떠오르면,
추가로 기록해야겠다.

 

 

Accessing RDMBS Using Twisted

Twsited Matrix Framework 상에서 DB에 접속하려면,
해당 DB의 python 모듈이 설치되어 있어야 한다.

예) mysql용 : mySQl for python 모듈 (sourceforge.net에서 다운받는다)

1. 필요한 라이브러리 import

from twisted.enterprise import adbapi from twisted.internet   import reactor

2. DB에 접속하기

dbpool = adbapi.ConnectionPool(‘dbmodule’, ‘mydb’, ‘id’, ‘password’)

의 형식으로 기록한다.

따라서, 다음의 두 방법과 같이 접속 정보를 기록할 수 있다.
  1) 방법 #1

dbpool = adbapi.ConnectionPool(‘MySQLdb’, db=’test’, user=’me’, passwd = ‘1234’)

  2) 방법 #2

DB_DRIVER = ‘MySQLdb’ DB_ARGS = {         ‘host’ : ‘localhost’,         ‘db’   : ‘test’,         ‘user’: ‘me’,         ‘passwd’ : ‘1234’, } dbpool = adbapi.Connectionpool(DB_DRIVER, **DB_ARGS)

3. 일반 query 실행시키기

def getCount():     return dbpool.runQuery(“SELECT * FROM log”) def printResult(l):     if l:         print l[0][0], ‘records’ getCount().addCallback(printResult)

위와 같이 하면, SELECT문이 asynchrous하게 실행되고, 결과가 도착하면 printResult 한다.

4. Callable  객체를 실행시키기 (함수, transaction)

def _get10(txn):     txn.execute(..do something..)     txn.execute(“SELECT * FROM log limit 10”)     result = txn.fetchall()          if result:         return result[0][0]     else:        return None def get10():     return dbpool.runInteraction(_get10) def print10(data):     if data != None:         print data, ‘data found’     else:         print ‘no such data’ get10().addCallback(print10)

이와 같이 하면, _get10()을 수행한 결과가 리턴된다.

5. 리턴값 형식 확인
리턴값이 result[0][0]의 형식으로 되는 이유는, 테이블이기 때문이다.
즉, 온전한 데이터는 x * y의 형태이다.

ㅁ len(result) = 결과값의 레코드 수 ㅁ len(result[0]) = 컬럼수 ㅁ len(result[0][0]) – 에러! 왜냐하면 2차원으로 참조하면 실질적인 데이터를    나타내기 때문이다.

Q) 자동으로 commit이 된다던데, 명시적으로 commit하거나 roll back 하는 함수도 있는지 찾아봐야겠다

OptionParser를 활용한 commandline parsing 하기

Optik (aka optarse) 를 사용하면, pytthon 애플리케이션 개발시 커맨드라인 옵션/
파라미터를 파싱하기 쉬워진다.

예를 들어서,

“실행파일명 –config 파일명” 으로 구성되는 형태의 옵션을 만들고 싶다면,
다음과 같이 간단하게 구성할 수 있다.

from optik import OptionParser def main():     usage = “usage: %prog [options] arg”     parser = OptionParser(usage)     parser.add_option(“-c”, “–config”, action=”store”, type=”string”, dest=”filename” )     (options, args) = parser.parse_args()         if options.filename:         print “cfg : %s” % options.filename main()

add_option() 내에 사용된 옵션에 대해 첨언하자면,
ㅁ -c : 한개의 대쉬 문자 뒤에 지정되는 문자는 1 캐릭터여야 한다
ㅁ –config : 풀로 표기하는 옵션은 두개의 대쉬 문자뒤에 표기한다.
   (두 개의 위치는 바뀌어도 상관없다. 표기만 제대로 한다면)
ㅁ action=”store” : c/config  옵션 뒤에 오는  argument를 활용하기 위해 지정함.
ㅁ type=”string” : 문자열을 받도록 해줌
ㅁ dest=”filename” : 나중에 c/config 등등 다양한 형태로 지정된 옵션을 통합하여,
  호출할 때 사용됨. 즉, c/config 옵션은 “filename”이라는 옵션으로 통합하여
  인지한다.
  action = “store” 로 저장한 파일이름은 options.filename의 속성값으로 활용한다.

기본적으로 -h/ –help 옵션은 deploy된다. 즉 테스트파일에 –help 옵션을 줘서
실행하면,

Usage: OptionTester.py [options] arg Options:   -h, –help            show this help message and exit   -c FILENAME, –config=FILENAME

와 같이 결과가 나온다.
여기에서 참고할 만한 내용은, 대쉬의 개수에 따라 동작이 다르게 될 수도 있다는 점.

동일한 결과를 내려면.

-c FILENAME –config=FILENAME –config FILENAME

의 형태로 활용할 수 있다.

보다 자세한 설명 및 활용 방법은
http://optik.sourceforge.net/ 에서 확인할 수 있다.