Still Playing With IRC and Erlang

I've been playing with the code I made for my poor man's erlang irc client, which was initially intended to be a bot, but mutated into a simple client, and now I'm making a GUI for it using wxErlang... you can see the code here: Ditto

Cheers,

Poor Man's Erlang IRC Client

I need comments on this.

-module(conn).
-import(lists, [concat/1, keysearch/3]).
-export([start/2, stop/0]).
-export([init/1, terminate/2, handle_event/2, handle_call/2, handle_info/2]). 
-export([repl/0, conn/3]).
-behaviour(gen_event).

conn(ClientPid, Host, PortNo) ->
    Result = gen_tcp:connect(Host, PortNo, [binary,
                                            {active, true},
                                            {packet, line},
                                            {keepalive, true},
                                            {nodelay, true}]),
    case Result of
        {ok, Socket} ->
            Pid = spawn(fun() -> loop(ClientPid, Socket) end),
            gen_tcp:controlling_process(Socket, Pid),
            Pid;
        {error, Reason} ->
            io:format("Error, reason: ~p~n", [Reason]),
            error
    end.

loop(ClientPid, Socket) ->
    receive
        {tcp, _Socket, Data} ->
            gen_event:notify(?MODULE, {recv, Data}), 
            loop(ClientPid, Socket);
        {tcp_closed, _Socket} ->
            io:format("[SOCKET] Connection Closed.~n"),
            ok;
        {tcp_error, _Socket, Reason} ->
            io:format("[SOCKET] Error, reason: ~p~n", [Reason]),
            ok;
        {send, ClientPid, Data} ->
            gen_tcp:send(Socket, Data),
            loop(ClientPid, Socket);
        {close, ClientPid} ->
            gen_tcp:close(Socket),
            io:format("[SOCKET] Closing connection...~n"),
            ok;
        Msg ->
            io:format("[SOCKET] Recv: ~p~n", [Msg])
    end.

%
% exported functions
%
start(Host, PortNo) ->
    gen_event:start({local, ?MODULE}),
    gen_event:add_handler(?MODULE, ?MODULE, [{host, Host}, {port, PortNo}]).

stop() ->
    gen_event:stop(?MODULE).


%
% helper
%
irc_msg(Args) ->
    lists:concat(Args) ++ "\r\n".

%
% gen_event behaviour
%
init(Args) ->
    io:format("*** initiating gen_event ***~n"),
    {value, {host, Host}} = keysearch(host, 1, Args),
    {value, {port, PortNo}} = keysearch(port, 1, Args),
    Pid = conn(self(), Host, PortNo),
    State = {disconnected, Pid},
    {ok, State}.

handle_event({recv, Line}, State) ->
    io:format("~s", [binary_to_list(Line)]),
    {ok, State};
handle_event({send, Line}, {_, Pid}=State) ->
    Pid ! {send, self(), Line ++ "\r\n"},
    {ok, State};
handle_event({pass, Password}, {disconnected, Pid}) ->
    Pid ! {send, self(), irc_msg(["PASS ", Password])},
    {ok, {nick, Pid}};
handle_event({nick, Nickname}, {nick, Pid}) ->
    Pid ! {send, self(), irc_msg(["NICK ", Nickname])},
    {ok, {user, Pid}};
handle_event({user, Nickname, Realname}, {user, Pid}) ->
    Pid ! {send, self(), irc_msg(["USER ", Nickname, " 0 * :", Realname])},
    {ok, {connected, Pid}};
handle_event({privmsg, To, Msg}, {connected, Pid}) ->
    Pid ! {send, self(), irc_msg(["PRIVMSG ", To, " :", Msg])},
    {ok, {connected, Pid}};
handle_event({join, Channel}, {connected, Pid}) ->
    Pid ! {send, self(), irc_msg(["JOIN ", Channel])},
    {ok, {connected, Pid}};
handle_event(quit, {connected, Pid}) ->
    Pid ! {send, self(), irc_msg(["QUIT"])},
    {ok, {disconnected, Pid}};
handle_event(Event, State) ->
    io:format("*** Error Unknow Event: ~p ***~n", [Event]),
    {ok, State}.

handle_call(_Request, State) ->
    io:format("~p~n", [State]),
    {ok, State, State}.

handle_info(_Info, State) ->
    io:format("~p, my pid: ~n", [State]),
    {ok, State}.

terminate({stop, Reason}, {_, Pid}) ->
    io:format("Stoping: ~p~n", [Reason]),
    Pid ! {close, self()},
    ok;
terminate(stop, {_, Pid}) ->
    io:format("Stoping.. ~n"),
    Pid ! {close, self()},
    ok.

%
% 
%

repl() ->
    io:format("*** Registering ***~n"),
    gen_event:notify(?MODULE, {pass, "some string"}),
    gen_event:notify(?MODULE, {nick, "mynickisdick"}),
    gen_event:notify(?MODULE, {user, "mynickisdick", "My Real Name"}),
    repl_loop("me:").

repl_loop(Prompt) ->
    case io:get_line(Prompt) of
        eof -> 
            ok;
        {error, Reason} ->
            io:format("error: ~w~n", [Reason]);
        Data ->
            CleanedData = string:strip(Data, both, $\n),
            Op = getops(CleanedData),
            case Op of
                {join, _Channel} ->
                    gen_event:notify(?MODULE, Op), 
                    repl_loop(Prompt);
                {privmsg, _To, _Msg} ->
                    gen_event:notify(?MODULE, Op),
                    repl_loop(Prompt);
                {quit} ->
                    gen_event:notify(?MODULE, quit);
                _ ->
                    io:format("Error, Op was: ~p~n", [Op]),
                    repl_loop(Prompt)
            end
    end.

getops(String) ->
    AvailOps = [{"/join", join, 1}, {"/quit", quit, 0}, {"/msg", privmsg, 2}],
    Msg = string:tokens(String, "\t\n "),
    Key = lists:nth(1, Msg),
    case keysearch(Key, 1, AvailOps) of
        {value, {"/msg", privmsg, 2}} ->
            {privmsg, lists:nth(2, Msg), 
             string:join(lists:nthtail(2, Msg), " ")}; 
        {value, {Key, Op, Len}} ->
            list_to_tuple([Op] ++ lists:sublist(Msg, 2, Len));
        _ ->
            Msg
    end.

How to use it?

$ erl
Erlang R13B01 (erts-5.7.2) [source] [smp:2:2] [rq:2] [async-threads:0] [kernel-poll:false]

Eshell V5.7.2  (abort with ^G)
1> c(conn).
./conn.erl:6: Warning: undefined callback function code_change/3 (behaviour 'gen_event')
{ok,conn}
2> conn:start("irc.freenode.org", 6667).
Initiating gen_event...
ok
3> conn:repl().
...

You got "/join #channel", "/msg #channel text to say" or "/msg nick test to say" and of course "/quit".

Unification Using Term-DAGs

What is unification? See here.

For the algorithm and formal explenation see here.

Why does the ConstantNode inherits from FunctionNode. That is because you can think ConstantNode as a Function with outdegree (arguments) 0.

What did you want to do with this? See here

# -*- coding: utf-8 -*-
import sys
from pprint import pprint


class TermNode(object):

    def __init__(self, symbol):
        self.symbol  = symbol
        self._parents = []
        self._children = []

    def _add_parent(self, node):
        """
        This function appends a parent to the node,
        parents should only be appended when we append self
        to another node.
        """
        self._parents.append(node)

    def empty_parents(self):
        """
        This function removes all parents
        """
        self._parents = []

    @property
    def parents(self):
        return self._parents

    @property
    def children(self):
        return self._children

    def _append_term(self, node):
        # check that it's a node, or at least has its interface
        if not hasattr(node, '_add_parent'):
            typename = str(type(node))
            raise TypeError('Node expected, instead ' + typename + ' given')

        self._children.append(node)
        node._add_parent(self)

    def __len__(self):
        return len(self._children)
    
    def __getitem__(self, index):
        if len(self) <= index:
            raise IndexError('index out of range')
        else:
            return self._children[index]

    def __setitem__(self, index, value):
        if len(self) <= index:
            raise IndexError('assignment index out of range')
        else:
            # value should be a node
            if not hasattr(value, '_add_parent'):
                typename = str(type(node))
                raise TypeError('Node expected, instead ' + typename + ' given')

            self._children[index] = value
            value._add_parent(self)

    def __contains__(self, other):
        stack = [self]

        while stack:
            elem = stack.pop(0)
            if other is elem:
                return True
            else:
                stack += elem.children
                
        return False

    def __repr__(self):
        return 'TermNode(' + repr(self.symbol) + ')'

    def is_function_node(self):
        return False

    def is_variable_node(self):
        return False

    def __eq__(self, other):
        return self.symbol == other.symbol

class VariableNode(TermNode):

    def is_variable_node(self):
        return True

    def __repr__(self):
        return "VariableNode(" + repr(self.symbol) + ")"


class FunctionNode(TermNode):

    def __init__(self, symbol, outdegree, children_list):
        if len(children_list) != outdegree:
            raise ValueError('The list of children is different'
                             ' than the outdegree')

        super(FunctionNode, self).__init__(symbol)
        self.outdegree = outdegree
        
        for child in children_list:
            self._append_term(child)

    def is_function_node(self):
        return True

    def __eq__(self, other):
        return (self.symbol == other.symbol and
                self.outdegree == other.outdegree)

    def __repr__(self):
        return ('FuntionNode(' + 
                repr(self.symbol) + ', '  +
                repr(self.outdegree) + ', ' +
                repr(self.children) + ')')


class ConstantNode(FunctionNode):

    def __init__(self, symbol):
        super(ConstantNode, self).__init__(symbol, 0, [])

    def __repr__(self):
        return ('ConstantNode(' +
                repr(self.symbol))
        
class SymbolClashException(Exception):
    pass

class OccursCheckException(Exception):
    pass

class TermDAG(object):
    """
    Represents a directe acyclic graph for terms
    """
    def replace(self, node, othernode):
        """
        Merges two node, say we have nodes s and t and we wan to merge them

        Let parents(s) = {p1, ..., pn }; then
        1. For each pi, replace the subterm arc pi -> s by pi -> t
            Note: this is done by p[k] = othernode in the code, where k is the
            position of the node 's' in the list and othernode is 't'
        2. Let parents(t) := parents(s) U parents(t);
            Note: U represents the set union; in code this is done
            when we you use __setitem__, that is p[k] = othernode
        3. Let parents(s) := Empty
        """

        for p in node.parents:
            for k, child in enumerate(p.children):
                if node is child:
                    p[k] = othernode
        node.empty_parents()

    def unify(self, node, other, sigma=None):
        if sigma is None:
            sigma = []
            
        if node is other:
            return

        elif node.is_function_node() and other.is_function_node():
            if node == other:
                for i in xrange(0, node.outdegree):
                    self.unify(node[i], other[i], sigma)
            else:
                raise SymbolClashException()
           
        elif not node.is_variable_node():
            self.unify(other, node, sigma)

        elif node in other:
            raise OccursCheckException()

        else:
            sigma.append((node, other))
            self.replace(node, other)
            
        return sigma

if __name__ == '__main__':
    y = VariableNode('y')

    f = FunctionNode('f', 2, [FunctionNode('g', 1, [y]),
                              FunctionNode('g', 1, [y]),
                              ])

    f_ = FunctionNode('f', 2, [VariableNode('x'),
                               FunctionNode('g', 1, [ConstantNode('a')])])

    print repr(f)
    print '--'

    print repr(f_)
    print '--'
    
    dag_manager = TermDAG()


    try:
        sigma = dag_manager.unify(f, f_)
    except SymbolClashException:
        print 'Failed to unify'
    except OccursCheckException:
        print 'Failed to unify'
    else:
        print sigma

Infinite list of primes! Yay!

or until the memory gives up...

# -*- coding: utf-8 -*-

from itertools import count
from collections import defaultdict

def seive():
    table = defaultdict(list)
    for x in count(2):
        facts = table[x]
        if facts:
            del table[x]
            for p in facts:
                table[x+p] = table[x+p] + [p]
        else:
            table[x*x] = [x]
            yield x
    
if __name__ == '__main__':
    for k, v in enumerate(seive()):
        print k, v

Got the idea from here.

This is an implementation of parallel dot product of two vectors. It seems obvious that one could use AMQP as messaging interface for parallel computing. So that's what I did here. I "ported" MPI Dot Product that was shown to me in a HPC class. The version of the algorithms is supposedly from Parallel Programming with MPI by Peter Pacheco.

I used RabbitMQ as the AMQP server and py-amqplib as client.

To use it, you should run a master: python dotprod.py -r 0 -n 2 and a worker:
python dotprod.py -r 1 -n 2. The -r option works like the "RANK" attribute of MPI. The -n option is how many processes are involved.

import sys

from cPickle import dumps, loads
from amqplib import client_0_8 as amqp
from optparse import OptionParser

my_rank = None
num_procs = None

DEBUG = True

# As there is no strict passing by reference I use "boxing" :-)
class Box(object):
    def __init__(self):
        self.data = None
        
    def pack(self, data):
        self.data = data

    def unpack(self):
        return self.data

    def __str__(self):
        return '%s' % self.data

def read_vector(conn, chan, prompt, vector_order, my_rank, num_procs):
    n_bar = vector_order / num_procs

    if (my_rank == 0):
        s = raw_input(("Enter a list of integers of size %d:" % vector_order))
        
        try:
            vector = [int(i) for i in s.split(",")] 

        except ValueError, e:
            print e
            chan.close()
            conn.close()
            sys.exit(1)

        local_vector = vector[0:n_bar] 

        fst, snd = n_bar, n_bar * 2
        for q in xrange(1, num_procs):
            slz = vector[fst:snd] 
            fst = snd
            snd += n_bar

            # Send
            msg = amqp.Message(dumps(slz))
            msg.properties['delivery_mode'] = 2
            chan.basic_publish(msg, exchange="sorting_room",
                               routing_key=("key_%d" % q))

        return local_vector
        
    else:
        # Receive
        local_data = Box()
        
        def recv_callback(msg):
            print ('Received, from channel #'
                   + str(msg.channel.channel_id)),
            local_data.pack(loads(msg.body))
            print 'data:', local_data
            
        chan.basic_consume(queue=('po_box_%d' % my_rank), no_ack=True,
                           callback=recv_callback,
                           consumer_tag=("tag_%d" % my_rank))
        chan.wait()
        chan.basic_cancel("tag_%d" % my_rank)

        return local_data.unpack()
        
        
def main(conn, chan, my_rank, num_procs):

    # Queue and exchange for broadcasting
    chan.exchange_declare(exchange="bcast", type="fanout", durable=True,
                          auto_delete=False)
    chan.queue_declare(queue=("bcast_queue_%d" % my_rank), durable=True,
                       exclusive=False, auto_delete=False)
    chan.queue_bind(queue=("bcast_queue_%d" % my_rank), exchange="bcast")

    # Queue and exchange for point-to-point communication
    chan.exchange_declare(exchange="sorting_room", type="direct",
                          durable=True, auto_delete=False)
    chan.queue_declare(queue=("po_box_%d" % my_rank), durable=True,
                           exclusive=False, auto_delete=False)
    chan.queue_bind(queue=("po_box_%d" % my_rank),
                    exchange="sorting_room",
                    routing_key=("key_%d" % my_rank))

    if (my_rank == 0):

        try:
            vector_order = int(raw_input("Enter an integer number:"))
        except ValueError, e:
            print e
            chan.close()
            conn.close()
            sys.exit(1)

        # Broadcasting to a set of programs
        msg = amqp.Message(dumps(vector_order))
        msg.properties["delivery_mode"] = 2
        chan.basic_publish(msg, exchange="bcast")

    else:
        # Receive Bcast
        local_data = Box()
        
        def recv_callback(msg, ):
            print ('Received, from channel #'
                   + str(msg.channel.channel_id)),
            print 'data:', loads(msg.body)
            local_data.pack(loads(msg.body))
   
        
        chan.basic_consume(queue=('bcast_queue_%d' % my_rank),
                           no_ack=True,
                           callback=recv_callback,
                           consumer_tag=("bcast_tag_%d" % my_rank))

        
        chan.wait()
        chan.basic_cancel("bcast_tag_%d" % my_rank)         
        print 'Done'

        vector_order = local_data.unpack()

    local_x = read_vector(conn, chan, "read first vector", vector_order,
                          my_rank, num_procs)
    local_y = read_vector(conn, chan, "read second vector", vector_order,
                          my_rank, num_procs)

    print 'Vectors:', local_x, local_y

    # local dot prod
    acc = 0.0
    for idx in xrange(0, len(local_x)):
        acc += local_x[idx] * local_y[idx]

    local_dot = acc

    print '[rank: %d] Local dot product: %f' % (my_rank, local_dot)

    if my_rank != 0:
        # send to master
        msg = amqp.Message(dumps(local_dot))
        msg.properties["delivery_mode"] = 2
        chan.basic_publish(msg, exchange="sorting_room",
                           routing_key="key_0")
        print '[rank: %d] done' % my_rank
    else:
        # Receive messages
        local_data = Box()
        
        def recv_callback(msg):
            print ('Received, from channel #'
                   + str(msg.channel.channel_id)),
            print 'data:', loads(msg.body)
            local_data.pack(loads(msg.body))
           
        chan.basic_consume(queue=('po_box_%d' % my_rank),
                           no_ack=True,
                           callback=recv_callback,
                           consumer_tag=("tag_%d" % my_rank))

        acc = 0.0
        for i in xrange(1, num_procs):
            chan.wait()
            acc += local_data.unpack()
            print '[rank: %d] acc: %f' % (my_rank, acc)

        chan.basic_cancel("tag_%d" % my_rank)         
        print '[rank: %d] done, result %f' % (my_rank, acc + local_dot)
    

if __name__ == '__main__':
    # Initializing connection
    conn = amqp.Connection(host="localhost:5672", userid="guest",
                           password="guest", virtual_host="/",
                           insist=False)
    chan = conn.channel()

    
    parser = OptionParser()
    parser.add_option("-r", "--rank", dest="rank",
                      help="Gives the rank to the process",
                      default=1)
    parser.add_option("-n", "--number_procs", dest="np", default=2,
                      help="Number of procs to use")

    (options, args) = parser.parse_args()

    try:
        my_rank = int(options.rank)
        num_procs = int(options.np)
    except ValueError, e:
        print e
        chan.close()
        conn.close()
        sys.exit(1)

    print 'Running with rank:', my_rank, 'and', num_procs, 'processes'

    main(conn, chan, my_rank, num_procs)

    # close channel and connection
    chan.close()
    conn.close()

Erlang Process Ring!

I though I might just post this. This is a couple of weeks old. It's the exercise from chapter four, regarding the erlang rings.

-module(ring).
-import(erlang, [error/1]).
-export([start/2]).
-export([init/0]).

-ifdef(debug).
-define(DBG(Str, Args), io:format(Str, Args).
-else.
-define(DBG(Str, Args), ok).
-endif.

%% Process that makes the ring
loop(Pid, Id) ->
    receive
 {stop, FromPid} ->
     ?DBG("~p received msg: stop from ~p\n", [self(), FromPid]);
 {Msg, FromPid} ->
     ?DBG("~p received msg: ~p from ~p\n", [self(), Msg, FromPid]),
     case Id of 
  first ->
      case Msg of 
   {_, 0} ->
       Pid ! {stop, self()};
   {Str, Num} ->
       Pid ! {{Str, Num - 1}, self()},
       loop(Pid, Id)     
      end;

  middle ->
      Pid ! {Msg, self()},
      loop(Pid, Id)
     end;

 Msg -> 
     io:format("wtf: ~p\n", [Msg]),
     error("bad message")
    end.


init() ->
    receive
 {next, Pid, _} -> 
     ?DBG("~p sends to ~p\n", [self(), Pid]),
     loop(Pid, middle);
 {first, Pid, Num} ->
            ?DBG("[first:~p] sends to ~p\n", [self(), Pid]),
            Pid ! {{"hello", Num}, self()},
     loop(Pid, first);
 Msg -> 
     io:format("wtf: ~p\n", [Msg]),
     error("bad init message")
    end,
    ok.

%% To set up the ring
create_procs(0, PidList) ->
    PidList;
create_procs(N, PidList) ->
    Pid = spawn(?MODULE, init, []),
    NewPidList = [Pid|PidList],
    create_procs(N - 1, NewPidList).

set_up_ring_(Num, Last, [Pid|[First|[]]]) -> %% Num is the number of messages
    Last ! {next, First, Num},
    ?DBG("~p <- ~p\n", [Last, First]),
    First ! {first, Pid, Num},
    ?DBG("~p <- ~p\n", [First, Pid]);
set_up_ring_(Num, Last, [Pid1|[Pid2|PidList]]) ->
    Pid2 ! {next, Pid1, Num},
    ?DBG("~p <- ~p\n", [Pid2, Pid1]),
    set_up_ring_(Num, Last, [Pid2|PidList]).
    
set_up_ring(Num, [X|[Y|XS]]) ->
    set_up_ring_(Num, X, [X|[Y|XS]]);
set_up_ring(_, _) ->
    error(badarg).

start(NumProcs, NumMessages) when (NumProcs >= 2) and (NumMessages >= 1) ->
    PidList = create_procs(NumProcs, []),
    set_up_ring(NumMessages, PidList).

Binary Trees in Erlang

More Erlang goodness, today binary trees!

The record which describes a node

-record(node, {left=nil, load, right=nil}).

Insertion and preorder

-module(binary_tree).
-export([newnode/1, insert/2, preorder/1]).
-import(lists).
-include("binary_tree.hrl").

newnode(Load) ->
    #node{load=Load}.

newleaf(Load) ->
    {leaf, Load}.

isnil(Node) ->
    Node == nil.

insert({leaf, LeafLoad}, Load) ->
    if 
        LeafLoad > Load ->
            #node{left=newleaf(Load), load=LeafLoad};
        LeafLoad =< Load ->
            #node{load=LeafLoad, right=newleaf(Load)}
    end;
insert(#node{left=LeftNode,
             load=CurrentLoad,
             right=RightNode} = Node,
       Load) ->
    if
        CurrentLoad > Load ->
            case isnil(LeftNode) of
                true ->
                    Node#node{left=newleaf(Load)};
                 false ->
                    Node#node{left=insert(LeftNode, Load)}
            end;
        CurrentLoad =< Load ->
            case isnil(RightNode) of
                true ->
                    Node#node{right=newleaf(Load)};
                false ->
                    Node#node{right=insert(RightNode, Load)}
            end
    end.

preorder([], ValueList) ->
    ValueList;
preorder([Tree|TreeList], ValueList) ->
    case Tree of
        #node{left=Left, load=Load, right=Right} ->
            NewValueList = [Load|ValueList],
            NewTreeList = [Left|[Right|TreeList]],
            preorder(NewTreeList, NewValueList);
        {leaf, LeafLoad} ->
            NewValueList = [LeafLoad|ValueList],
            preorder(TreeList, NewValueList);
        nil ->
            preorder(TreeList, ValueList)
    end.
           
preorder(Tree) ->
    case Tree of
        #node{left=_Left, load=_Load, right=_Right} ->
            lists:reverse(preorder([Tree], []));
        nil ->
            []
    end.

Weeee....