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教程