使用Python扩展到Erlang

作者: 日期:

gashero

2008-04-21

目录

• 1   Erlang服务器
  • 1.1   启动服务器
  • 1.2   停止服务器
  • 1.3   发送查询请求
  • 1.4   Port主循环
  • 1.5   消息的收发流程
  • 1.6   被切分行的重新组装
• 2   Python客户端

Erlang可以通过stdin/stdout与其他语言编写的进程交互，实现程序功能的扩展。这个例子使用了Python来扩展Erlang的功能，完善的演示了Erlang做扩展的方式和一些必要的工具函数。

这个例子使用line模式。

1   Erlang服务器

1.1   启动服务器

启动一个进程来连接到port，然后循环等待查询消息。

start() ->
    spawn(fun() ->
            register(expy,self()),
            process_flag(trap_exit,true),
            Port=open_port({spawn,"python -u client.py"},[{line,8}]),
            portloop(Port)
        end).

这里启动了一个进程，并且注册到名字expy。启动的命令为 python -u client.py 。

Note

程序的IO缓存，有些应用程序会对IO进行缓存处理，导致双方无法及时收到消息。对Python来说，可以用 -u 启动参数来关闭IO缓存。或者在程序中每次输出之后立即调用stdout的flush()方法。

1.2   停止服务器

只要给循环等待查询请求的进程发送stop消息即可，如下:

stop() ->
    expy ! stop,
    ok.

1.3   发送查询请求

因为本次例子以行模式运行，于是发送以行为单位的请求，发送到expy进程:

callline(Line) ->
    expy ! {call,{self(),Line++"\n"}},
    receive
        {reply,Data} ->
            Data;
        noreply ->
            io:format("callport no reply!~n");
        Other ->
            io:format("callport Other: ~p~n",[Other])
    end.

1.4   Port主循环

用于将查询请求转发给Port程序，接受port响应等。

portloop(Port) ->
    %io:format("PortLoop ready!~n"),
    receive
        {call,{Caller,Msg}} ->
            Port ! {self(),{command,Msg}},
            io:format("Sent message to python~n"),
            receive
                {Port,{data,{eol,Data}}} ->
                    %io:format("Reply: ~p~n",[Data]);
                    Caller ! {reply,Data},
                    portloop(Port);
                {Port,{data,{noeol,Part}}} ->   %会继续接受消息，直到最后一个是eol
                    %io:format("ReplyPart: ~p~n",[Part]);
                    io:format("long line~n"),
                    Line=recvlongline(Port,Part),
                    Caller ! {reply,Line},
                    portloop(Port);
                {'EXIT',Port,Reason} ->
                    io:format("Port exited: ~p~n",[Reason]),
                    unregister(expy),
                    Caller ! noreply,
                    exit(normal);
                Other ->
                    io:format("portloop 2 Other: ~p~n",[Other]),
                    Caller ! noreply
            end;
        stop ->
            Port ! {self(),close},
            receive
                {Port,closed} ->
                    unregister(expy),
                    exit(normal)
            end;
        {'EXIT',Port,Reason} ->
            io:format("Port exited: ~p~n",[Reason]),
            unregister(expy),
            %Caller ! noreply,
            exit(normal);
        Other ->
            io:format("portloop 1 Other: ~p~n",[Other])
            %portloop(Port)
    end.

顶层接受的消息的意义：

1. {call,{Caller,Msg}} ：查询请求，在循环内转发给Port之后等待Port的响应消息，并将结果发回给原来的调用者。
2. stop ：停止服务器，接受消息后停止服务器。
3. {'EXIT',Port,Reason} ：检测到Port主动停止的消息。
4. Other ：防止编程错误导致的错误消息无法捕捉。

1.5   消息的收发流程

向Port发送消息实际上是通过 Port ! {self(),{command,Msg}} 实现的。这样就把一行数据发送到了Port。

随后需要等待Port的响应消息，响应消息分为两种：

1. {Port,{data,{eol,Data}}} ：收到了完整的响应行数据，或者对于多块的响应，收到了最后一块。
2. {Port,{data,{noeol,Data}}} ：收到了不完整的一块数据。有后续数据需要继续等待消息。

因为行模式指定了最大一行允许接受的字符数量，那么在一行数据的长度超过这个数字时，就会导致消息被切分。每个收到的消息都是noeol类型，而最后一段完成整行的消息则是eol类型。

作为一种异常状况，应该考虑到行被切分的可能。

另外，就是Port可能会提前退出，这时会收到消息 {'EXIT',Port,Reason} 。可以自己定义处理，不过一般都需要注销对应进程，发送友好的回应，并且把自身进程结束掉。

1.6   被切分行的重新组装

一般按照需要决定是否还要组装对应的报文，所以这里定义了两个函数来接受超长的报文行，一种是接受并组装，另一种是接收后丢弃:

%% 持续接受超长的行
recvlongline(Port,LastLine) ->
    receive
        {Port,{data,{noeol,Data}}} ->
            %io:format("long part: ~p~n",[Data]),
            recvlongline(Port,LastLine ++ Data);
        {Port,{data,{eol,Data}}} ->
            LastLine ++ Data
    end.

%% 丢弃超长的行
droplongline(Port) ->
    receive
        {Port,{data,{noeol,_Data}}} ->
            droplongline(Port);
        {Port,{data,{eol,_Data}}} ->
            ok
    end.

2   Python客户端

客户端的 stdin 用于读取Erlang发来的命令，而 stdout 用于发送响应到Erlang。这个时候剩余的 stderr 可以用于打印自定义的错误消息，方便调试。

如下是完整的程序:

#! /usr/bin/env python
# -*- coding: UTF-8 -*-
# File: client.py
# Date: 2008-04-16
# Author: harry

"""
测试erlang与python的port
"""

import os
import sys
import time
import traceback

def process(line):
    try:
        bb=eval(line)
        return bb
    except:
        return repr(traceback.format_exc())

def lineio():
    while True:
        #print >> sys.stderr,'--'
        line=sys.stdin.readline().strip()
        #print >> sys.stderr, "Line:"+line
        if not line:
            #print >> sys.stderr, "break out"
            break
        if line=='exit':
            break
        #print >> sys.stderr,'++'
        reply=process(line)
        sys.stdout.write(str(reply)+"\n")
        sys.stdout.flush()
        #print >> sys.stdout,str(reply)+"\n"    #XXX: 这种不行
    return

def main():
    #print >> sys.stderr, 'python start!'
    lineio()
    return

if __name__=='__main__':
    main()

其中需要注意的是：

1. 一般来说Port程序都是处于循环状态，接收Erlang发来的请求，处理并响应。而不应该处理完成后直接退出。
2. 接收请求行时会得到行为有换行符的请求，注意去掉。
3. 可以通过 stderr 打印错误调试信息。
4. 发送响应报文时末尾要加上换行符。
5. 响应报文本身注意对包含换行符情况的处理。
6. 如果IO有缓存，那么可以选择关闭缓存，或者在写入响应后立即flush()。

评论

1 楼 upsuper 2010-04-06  

python里可以使用 os.read 和 os.write 来规避一切缓存