Erlang/Elixir: 外部通讯之-C节点

时间  2019-12-14
标签 erlang elixir 外部 通讯 节点 栏目 C&C++ 繁體版
原文   原文链接

系列:
Erlang/Elixir: 外部通讯之-NIF
Erlang/Elixir: 外部通讯之-端口驱动
Erlang/Elixir: 外部通讯之-C节点css

C节点使用Erlang提供的 Erl_Interface 与 Erlang VM进行交互, 所以须要在C文件中包含头文件:
#include "erl_interface.h"html

2016-08-14 更新: C节点多线程示例 https://github.com/apofiget/c...node

从Erlang的角度看, C节点就像一个普通的Erlang节点同样. 调用C节点中的foo和bar函数就是给C节点发送消息, 并接收执行结果. 发送消息要求指定一个接收者, 接收者是一个Erlang进程ID, 或由一个元组({RegName, Node})表示的进程.git

若是不知道PID, 那么能够经过下面的方式给接收者发送消息.github

{RegName, Node} ! Msg

节点名称 Node 为C节点的名称. 若是节点使用短名称, 必须遵循 cN这种命名模式, N为整数shell

C端

在调用 Erl_Interface接口中的其余函数前, 须要初始化内存.segmentfault

erl_init(NULL, 0);

如今就能够初始化C节点了. 若是使用短节点名称, 经过调用 erl_connect_init() 完成节点的初始化:服务器

erl_connect_init(1, "secretcookie", 0);

其中:cookie

  • 第一个参数为整数, 用于构造节点名称, 此例中节点名称为 c1多线程

  • 第二个参数为字符串, 用于设置Cookie的值

  • 第三个参数为一个整数, 用于标识一个C节点实例.

若是使用长名称, 须要调用 erl_connect_xinit() 进行初始化, 而不是 erl_connect_init():

erl_connect_xinit(
    "idril", "cnode",    "cnode@idril.du.uab.ericsson.se", &addr,  "secretcookie", 0
);
----------------------------------------------------------------------------------------------
    主机名称  本地节点名称  全称                               地址     Cookie值        实例编号

其中:

  • 第一个参数为主机名称

  • 第二个参数为节点本地名称(不包含域名部分)

  • 第三个参数为节点的全称

  • 第四个参数为一个指针, 指向一个包含该主机IP地址的 in_addr 结构.

  • 第五个参数为Cookie的值

  • 第六个参数为实例编号

C节点在设置Erlang和C之间的通讯的时候, 既能够做为服务器端, 也能够做为客户端. 若是做为客户端, 须要经过调用 erl_connect() 链接到Erlang节点, 成功链接后, 返回一个打开的文件描述符:

fd = erl_connect("e1@localhost");

若是C端做为一个服务器运行, 它必须首先建立一个套接字(调用bind()和listen())来监听特定的端口. 而后把名称和端口发布到epmd(Erlang端口映射守护进程), 详细信息请参考 手册.

erl_publish(port);

如今C节点服务器能够接受来自Erlang节点的链接了.

fd = erl_accept(listen, &conn);

erl_accept 的第二个参数为一个 ErlConnect 结构, 包含链接相关的信息. 例如, Erlang节点的名字.

收发消息

C节点能够调用 erl_receive_msg() 接收来自 Erlang节点的消息. 该函数从一个打开的文件描述符fd中读取数据, 并复制到一个缓冲区(Buffer)中, 接收的消息被存放在名为 ErlMessage 的结构 emsg 中. ErlMessage 的 type 字段代表接收的消息的类型. ERL_REG_SEND 指出, Erlang发送了一条消息到C节点中的一个已注册进程. 实际的消息是一个 ETERM, 在 ErlMessage结构的 msg 字段中.

节点事件

  • ERL_ERROR 发生了错误

  • ERL_TICK 节点心跳

  • link

  • unlink

  • exit

节点心跳事件(ERL_TICK)应该被忽略或输出到日志, 错误事件应该被处理

while (loop) {

    got = erl_receive_msg(fd, buf, BUFSIZE, &emsg);
    if (got == ERL_TICK) {
      /* ignore */
    } else if (got == ERL_ERROR) {
      loop = 0; /* exit while loop */
    } else {
      if (emsg.type == ERL_REG_SEND) {

由于消息是一个 ETERM 结构, Erl_Interface 接口中的函数操做. 在这个例子中, 消息体为一个三元组,第二个元素为调用者的pid,第三个元素为元组 {Function,Arg}, 用于决定要调用的函数. 函数的执行结果被封装成一个 ETERM 结构并调用 erl_send() 函数返回给调用者, 它接受三个参数, 分别是: 文件描述符, Pid, 以及一个项式:

fromp = erl_element(2, emsg.msg);
    tuplep = erl_element(3, emsg.msg);
    fnp = erl_element(1, tuplep);
    argp = erl_element(2, tuplep);

    if (strncmp(ERL_ATOM_PTR(fnp), "foo", 3) == 0) {
      res = foo(ERL_INT_VALUE(argp));
    } else if (strncmp(ERL_ATOM_PTR(fnp), "bar", 3) == 0) {
      res = bar(ERL_INT_VALUE(argp));
    }

    resp = erl_format("{cnode, ~i}", res);
    erl_send(fd, fromp, resp);

最后, 经过 ETERM 建立函数函数分配的内存必须被释放(包括经过erl_receive_msg()函数建立的)

erl_free_term(emsg.from); erl_free_term(emsg.msg);
    erl_free_term(fromp); erl_free_term(tuplep);
    erl_free_term(fnp); erl_free_term(argp);
    erl_free_term(resp);

下面是一个使用短节点名称的C节点服务器实现

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include "erl_interface.h"
#include "ei.h"
#include "complex.h"
#include "listen.h"

#define BUFSIZE 1000

int main(int argc, char **argv) {
  int port;                                /* Listen port number */
  int listen;                              /* Listen socket */
  int fd;                                  /* fd to Erlang node */
  ErlConnect conn;                         /* Connection data */

  int loop = 1;                            /* Loop flag */
  int got;                                 /* Result of receive */
  unsigned char buf[BUFSIZE];              /* Buffer for incoming message */
  ErlMessage emsg;                         /* Incoming message */

  ETERM *fromp, *tuplep, *fnp, *argp, *resp;
  int res;

  port = atoi(argv[1]);

  erl_init(NULL, 0);

  if (erl_connect_init(1, "secretcookie", 0) == -1)
    erl_err_quit("erl_connect_init");

  /* Make a listen socket */
  if ((listen = my_listen(port)) <= 0)
    erl_err_quit("my_listen");

  // Publish to epmd
  if (erl_publish(port) == -1){
    erl_err_quit("erl_publish");
  }

  if ((fd = erl_accept(listen, &conn)) == ERL_ERROR){
    erl_err_quit("erl_accept");
  }
  fprintf(stderr, "Connected to %s\n\r", conn.nodename);

  while (loop) {

    got = erl_receive_msg(fd, buf, BUFSIZE, &emsg);
    if (got == ERL_TICK) {
      /* ignore */
    } else if (got == ERL_ERROR) {
      loop = 0;
    } else {

      if (emsg.type == ERL_REG_SEND) {
        fromp = erl_element(2, emsg.msg);
        tuplep = erl_element(3, emsg.msg);
        fnp = erl_element(1, tuplep);
        argp = erl_element(2, tuplep);

        if (strncmp(ERL_ATOM_PTR(fnp), "foo", 3) == 0) {
          res = foo(ERL_INT_VALUE(argp));
        } else if (strncmp(ERL_ATOM_PTR(fnp), "bar", 3) == 0) {
          res = bar(ERL_INT_VALUE(argp));
        }

        resp = erl_format("{cnode, ~i}", res);
        erl_send(fd, fromp, resp);

        erl_free_term(emsg.from); erl_free_term(emsg.msg);
        erl_free_term(fromp); erl_free_term(tuplep);
        erl_free_term(fnp); erl_free_term(argp);
        erl_free_term(resp);
      }
    }
  } /* while */
}


int my_listen(int port) {
  int listen_fd;
  struct sockaddr_in addr;
  int on = 1;

  if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0){
    return (-1);
  }

  setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));

  memset((void*) &addr, 0, (size_t) sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_port = htons(port);
  addr.sin_addr.s_addr = htonl(INADDR_ANY);

  if (bind(listen_fd, (struct sockaddr*) &addr, sizeof(addr)) < 0){
    return (-1);
  }else{
    printf("server is listen on: %d\n", port);
  }

  listen(listen_fd, 5);
  return listen_fd;
}

下面是一个使用长节点名的C节点服务器实现

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include "erl_interface.h"
#include "ei.h"
#include "complex.h"
#include "listen.h"

#define BUFSIZE 1000

int main(int argc, char **argv) {
  struct in_addr addr;                     /* 32-bit IP number of host */
  int port;                                /* Listen port number */
  int listen;                              /* Listen socket */
  int fd;                                  /* fd to Erlang node */
  ErlConnect conn;                         /* Connection data */

  int loop = 1;                            /* Loop flag */
  int got;                                 /* Result of receive */
  unsigned char buf[BUFSIZE];              /* Buffer for incoming message */
  ErlMessage emsg;                         /* Incoming message */

  ETERM *fromp, *tuplep, *fnp, *argp, *resp;
  int res;

  port = atoi(argv[1]);

  erl_init(NULL, 0);

  addr.s_addr = inet_addr("134.138.177.89");
  if (erl_connect_xinit("idril", "cnode", "cnode@idril.du.uab.ericsson.se",
      &addr, "secretcookie", 0) == -1)
    erl_err_quit("erl_connect_xinit");

  /* Make a listen socket */
  if ((listen = my_listen(port)) <= 0)
    erl_err_quit("my_listen");

  if (erl_publish(port) == -1)
    erl_err_quit("erl_publish");

  if ((fd = erl_accept(listen, &conn)) == ERL_ERROR)
    erl_err_quit("erl_accept");
  fprintf(stderr, "Connected to %s\n\r", conn.nodename);

  while (loop) {

    got = erl_receive_msg(fd, buf, BUFSIZE, &emsg);
    if (got == ERL_TICK) {
      /* ignore */
    } else if (got == ERL_ERROR) {
      loop = 0;
    } else {

      if (emsg.type == ERL_REG_SEND) {
        fromp = erl_element(2, emsg.msg);
        tuplep = erl_element(3, emsg.msg);
        fnp = erl_element(1, tuplep);
        argp = erl_element(2, tuplep);

        if (strncmp(ERL_ATOM_PTR(fnp), "foo", 3) == 0) {
          res = foo(ERL_INT_VALUE(argp));
        } else if (strncmp(ERL_ATOM_PTR(fnp), "bar", 3) == 0) {
          res = bar(ERL_INT_VALUE(argp));
        }

        resp = erl_format("{cnode, ~i}", res);
        erl_send(fd, fromp, resp);

        erl_free_term(emsg.from); erl_free_term(emsg.msg);
        erl_free_term(fromp); erl_free_term(tuplep);
        erl_free_term(fnp); erl_free_term(argp);
        erl_free_term(resp);
      }
    }
  }
}


int my_listen(int port) {
  int listen_fd;
  struct sockaddr_in addr;
  int on = 1;

  if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
    return (-1);

  setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));

  memset((void*) &addr, 0, (size_t) sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_port = htons(port);
  addr.sin_addr.s_addr = htonl(INADDR_ANY);

  if (bind(listen_fd, (struct sockaddr*) &addr, sizeof(addr)) < 0)
    return (-1);

  listen(listen_fd, 5);
  return listen_fd;
}

最后是C节点客户端代码实现

#include <stdio.h>
#include <string.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>

#include "erl_interface.h"
#include "ei.h"
#include "complex.h"

#define BUFSIZE 1000

int main(int argc, char **argv) {
  int fd;                                  /* fd to Erlang node */
  int loop = 1;                            /* Loop flag */
  int got;                                 /* Result of receive */
  unsigned char buf[BUFSIZE];              /* Buffer for incoming message */
  ErlMessage emsg;                         /* Incoming message */
  ETERM *fromp, *tuplep, *fnp, *argp, *resp;
  int res;
  erl_init(NULL, 0);
  if (erl_connect_init(1, "secretcookie", 0) == -1){
    erl_err_quit("erl_connect_init");
  }
  if ((fd = erl_connect("e1@localhost")) < 0){
    erl_err_quit("erl_connect");
  }
  fprintf(stderr, "Connected to e1@localhost\n\r");
  while (loop) {
    got = erl_receive_msg(fd, buf, BUFSIZE, &emsg);
    if (got == ERL_TICK) {
      /* ignore */
    } else if (got == ERL_ERROR) {
      loop = 0;
    } else {
      if (emsg.type == ERL_REG_SEND) {
        fromp  = erl_element(2, emsg.msg);
        tuplep = erl_element(3, emsg.msg);
        fnp    = erl_element(1, tuplep);
        argp   = erl_element(2, tuplep);
        if (strncmp(ERL_ATOM_PTR(fnp), "foo", 3) == 0) {
          res = foo(ERL_INT_VALUE(argp));
        } else if (strncmp(ERL_ATOM_PTR(fnp), "bar", 3) == 0) {
          res = bar(ERL_INT_VALUE(argp));
        }
        resp = erl_format("{cnode, ~i}", res);
        erl_send(fd, fromp, resp);
        erl_free_term(emsg.from); erl_free_term(emsg.msg);
        erl_free_term(fromp); erl_free_term(tuplep);
        erl_free_term(fnp); erl_free_term(argp);
        erl_free_term(resp);
      }
    }
  }
}

本文中的源码通过修改, 遵循C99标准.

运行这个例子

图片描述

下面是两个节点启动后, EPMD的注册名称

图片描述

启动服务器(短名称)

./bin/c_node_server 3456

启动Erlang节点

# 进入src目录

cd src

# 编译

erlc *.erl

# 启动节点并调用C节点的函数

➜  src erl -sname e1 -setcookie secretcookie

Eshell V7.3  (abort with ^G)
(e1@localhost)1> c_node_short:bar(4).
Result: 8
ok
(e1@localhost)2> c_node_short:bar(5).
Result: 10
ok
(e1@localhost)3>

代码库

https://github.com/developerw...

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程