erl_bif_port.c

来自「OTP是开放电信平台的简称」· C语言 代码 · 共 763 行 · 第 1/2 页

    prt = id_or_name2port(BIF_P, portid);
    if (!prt) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if (prt->bp != NULL) {
	free_message_buffer(prt->bp);
	prt->bp = NULL;
    }
    if (IS_CONST(data)) {
	prt->data = data;
    } else {
	Uint size;
	ErlHeapFragment* bp;
	Eterm* hp;

	size = size_object(data);
	prt->bp = bp = new_message_buffer(size);
	hp = bp->mem;
	prt->data = copy_struct(data, size, &hp, &bp->off_heap);
    }
    erts_smp_port_unlock(prt);
    BIF_RET(am_true);
}


BIF_RETTYPE port_get_data_1(BIF_ALIST_1)
{
    BIF_RETTYPE res;
    Port* prt;
    Eterm portid = BIF_ARG_1;

    prt = id_or_name2port(BIF_P, portid);
    if (!prt) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if (prt->bp == NULL) {	/* MUST be CONST! */
	res = prt->data;
    } else {
	Eterm* hp = HAlloc(BIF_P, prt->bp->size);
	res = copy_struct(prt->data, prt->bp->size, &hp, &MSO(BIF_P));
    }
    erts_smp_port_unlock(prt);
    BIF_RET(res);
}

/* 
 * Open a port. Most of the work is not done here but rather in
 * the file io.c.
 * Error returns: -1 or -2 returned from open_driver (-2 implies
 * 'errno' contains error code; -1 means we don't really know what happened),
 * -3 if argument parsing failed.
 */
static int
open_port(Process* p, Eterm name, Eterm settings, int *err_nump)
{
#define OPEN_PORT_ERROR(VAL) do { port_num = (VAL); goto do_return; } while (0)
    int i, port_num;
    Eterm option;
    Uint arity;
    Eterm* tp;
    Uint* nargs;
    ErlDrvEntry* driver;
    char* name_buf = NULL;
    SysDriverOpts opts;
    int binary_io;
    int soft_eof;
    Sint linebuf;
    byte dir[MAXPATHLEN];

    /* These are the defaults */
    opts.packet_bytes = 0;
    opts.use_stdio = 1;
    opts.redir_stderr = 0;
    opts.read_write = 0;
    opts.hide_window = 0;
    opts.wd = NULL;
    opts.envir = NULL;
    opts.exit_status = 0;
#ifdef _OSE_
    opts.process_type = OS_BG_PROC;
    opts.priority = 20;
#endif
    binary_io = 0;
    soft_eof = 0;
    linebuf = 0;

    *err_nump = 0;

    if (is_not_list(settings) && is_not_nil(settings))
	OPEN_PORT_ERROR(-3);

    /*
     * Parse the settings.
     */

    if (is_not_nil(settings)) {
	nargs = list_val(settings);
	while (1) {
	    if (is_tuple(*nargs)) {
		tp = tuple_val(*nargs);
		arity = *tp++;
		if (arity != make_arityval(2))
		    OPEN_PORT_ERROR(-3);
		option = *tp++;
		if (option == am_packet) {
		   if (is_not_small(*tp))
		      OPEN_PORT_ERROR(-3);
		   opts.packet_bytes = signed_val(*tp);
		   switch (opts.packet_bytes) {
		    case 1:
		    case 2:
		    case 4:
		      break;
		    default:
		      OPEN_PORT_ERROR(-3);
		   }
		} else if (option == am_line) {
		    if (is_not_small(*tp))
			OPEN_PORT_ERROR(-3);
		    linebuf = signed_val(*tp);
		    if(linebuf <= 0)
			OPEN_PORT_ERROR(-3);
		} else if (option == am_env) {
		    byte* bytes;
		    if ((bytes = convert_environment(p, *tp)) == NULL) {
			OPEN_PORT_ERROR(-3);
		    }
		    opts.envir = (char *) bytes;
		} else if (option == am_cd) {
		    Eterm iolist;
		    Eterm heap[4];
		    int r;

		    heap[0] = *tp;
		    heap[1] = make_list(heap+2);
		    heap[2] = make_small(0);
		    heap[3] = NIL;
		    iolist = make_list(heap);
		    r = io_list_to_buf(iolist, (char*) dir, MAXPATHLEN);
		    if (r < 0) {
			OPEN_PORT_ERROR(-3);
		    }
		    opts.wd = (char *) dir;
		} else {
		   OPEN_PORT_ERROR(-3);
	       }
	    } else if (*nargs == am_stream) {
		opts.packet_bytes = 0;
	    } else if (*nargs == am_use_stdio) {
		opts.use_stdio = 1;
	    } else if (*nargs == am_stderr_to_stdout) {
		opts.redir_stderr = 1;
	    } else if (*nargs == am_line) {
		linebuf = 512;
	    } else if (*nargs == am_nouse_stdio) {
		opts.use_stdio = 0;
	    } else if (*nargs == am_binary) {
		binary_io = 1;
	    } else if (*nargs == am_in) {
		opts.read_write |= DO_READ;
	    } else if (*nargs == am_out) {
		opts.read_write |= DO_WRITE;
	    } else if (*nargs == am_eof) {
		soft_eof = 1;
	    } else if (*nargs == am_hide) {
		opts.hide_window = 1;
	    } else if (*nargs == am_exit_status) {
		opts.exit_status = 1;
	    } 
#ifdef _OSE_
	    else if (option == am_ose_process_type) {
	      if (is_not_atom(*tp))
		OPEN_PORT_ERROR(-3);
	      if (*tp == am_ose_pri_proc)      opts.process_type = OS_PRI_PROC;
	      else if (*tp == am_ose_int_proc) opts.process_type = OS_INT_PROC;
	      else if (*tp == am_ose_bg_proc)  opts.process_type = OS_BG_PROC;
	      else if (*tp == am_ose_ti_proc)  opts.process_type = OS_TI_PROC;
	      else if (*tp == am_ose_phantom)  opts.process_type = OS_PHANTOM;
	      else OPEN_PORT_ERROR(-3);
	    } 
	    else if (option == am_ose_process_prio) {
	      if (is_not_small(*tp))
		  OPEN_PORT_ERROR(-3);
	      opts.priority = signed_val(*tp);
	      if((opts.priority <= 0) || (opts.priority > 31))
		  OPEN_PORT_ERROR(-3);		
	    }  
#endif
	    else {
		OPEN_PORT_ERROR(-3);
	    }
	    if (is_nil(*++nargs)) 
		break;
	    if (is_not_list(*nargs)) 
		OPEN_PORT_ERROR(-3);
	    nargs = list_val(*nargs);
	}
    }
    if (opts.read_write == 0)	/* implement default */
	opts.read_write = DO_READ|DO_WRITE;

    /* Mutually exclusive arguments. */
    if((linebuf && opts.packet_bytes) || 
       (opts.redir_stderr && !opts.use_stdio))
	OPEN_PORT_ERROR(-3); 

    /*
     * Parse the first argument and start the appropriate driver.
     */

    
    if (is_atom(name) || (i = is_string(name))) {
	/* a vanilla port */
	if (is_atom(name)) {
	    name_buf = (char *) erts_alloc(ERTS_ALC_T_TMP,
					   atom_tab(atom_val(name))->len+1);
	    sys_memcpy((void *) name_buf,
		       (void *) atom_tab(atom_val(name))->name, 
		       atom_tab(atom_val(name))->len);
	    name_buf[atom_tab(atom_val(name))->len] = '\0';
	} else {
	    name_buf = (char *) erts_alloc(ERTS_ALC_T_TMP, i + 1);
	    if (intlist_to_buf(name, name_buf, i) != i)
		erl_exit(1, "%s:%d: Internal error\n", __FILE__, __LINE__);
	    name_buf[i] = '\0';
	}
	driver = &vanilla_driver_entry;
    } else {   
	if (is_not_tuple(name))
	    OPEN_PORT_ERROR(-3);		/* Not a process or fd port */
	tp = tuple_val(name);
	arity = *tp++;

	if (*tp == am_spawn) {	/* A process port */
	    if (arity != make_arityval(2)) {
		OPEN_PORT_ERROR(-3);
	    }
	    name = tp[1];
	    if (is_atom(name)) {
		name_buf = (char *) erts_alloc(ERTS_ALC_T_TMP,
					       atom_tab(atom_val(name))->len+1);
		sys_memcpy((void *) name_buf,
			   (void *) atom_tab(atom_val(name))->name, 
			   atom_tab(atom_val(name))->len);
		name_buf[atom_tab(atom_val(name))->len] = '\0';
	    } else  if ((i = is_string(name))) {
		name_buf = (char *) erts_alloc(ERTS_ALC_T_TMP, i + 1);
		if (intlist_to_buf(name, name_buf, i) != i)
		    erl_exit(1, "%s:%d: Internal error\n", __FILE__, __LINE__);
		name_buf[i] = '\0';
	    } else
		OPEN_PORT_ERROR(-3);
	    driver = &spawn_driver_entry;
	} else if (*tp == am_fd) { /* An fd port */
	    int n;
	    struct Sint_buf sbuf;
	    char* p;

	    if (arity != make_arityval(3)) {
		OPEN_PORT_ERROR(-3);
	    }
	    if (is_not_small(tp[1]) || is_not_small(tp[2])) {
		OPEN_PORT_ERROR(-3);
	    }
	    opts.ifd = unsigned_val(tp[1]);
	    opts.ofd = unsigned_val(tp[2]);

	    /* Syntesize name from input and output descriptor. */
	    name_buf = erts_alloc(ERTS_ALC_T_TMP,
				  2*sizeof(struct Sint_buf) + 2); 
	    p = Sint_to_buf(opts.ifd, &sbuf);
	    n = sys_strlen(p);
	    sys_strncpy(name_buf, p, n);
	    name_buf[n] = '/';
	    p = Sint_to_buf(opts.ofd, &sbuf);
	    sys_strcpy(name_buf+n+1, p);

	    driver = &fd_driver_entry;
	} else
	    OPEN_PORT_ERROR(-3);
    }

    if (driver != &spawn_driver_entry && opts.exit_status) {
	OPEN_PORT_ERROR(-3);
    }

    port_num = erts_open_driver(driver, p->id, name_buf, &opts, err_nump);
    if (port_num < 0) {
	DEBUGF(("open_driver returned %d(%d)\n", port_num, *err_nump));
	OPEN_PORT_ERROR(port_num);
    }

    if (binary_io) {
	erts_port_status_bor_set(&erts_port[port_num],
				 ERTS_PORT_SFLG_BINARY_IO);
    }
    if (soft_eof) {
	erts_port_status_bor_set(&erts_port[port_num],
				 ERTS_PORT_SFLG_SOFT_EOF);
    }
    if (linebuf && erts_port[port_num].linebuf == NULL){
	erts_port[port_num].linebuf = allocate_linebuf(linebuf); 
	erts_port_status_bor_set(&erts_port[port_num],
				 ERTS_PORT_SFLG_LINEBUF_IO);
    }

 do_return:
    if (name_buf)
	erts_free(ERTS_ALC_T_TMP, (void *) name_buf);
    return port_num;
#undef OPEN_PORT_ERROR
}

static byte* convert_environment(Process* p, Eterm env)
{
    Eterm all;
    Eterm* temp_heap;
    Eterm* hp;
    Uint heap_size;
    int n;
    byte* bytes;

    if ((n = list_length(env)) < 0) {
	return NULL;
    }
    heap_size = 2*(5*n+1);
    temp_heap = hp = (Eterm *) erts_alloc(ERTS_ALC_T_TMP, heap_size*sizeof(Eterm));
    bytes = NULL;		/* Indicating error */

    /*
     * All errors below are handled by jumping to 'done', to ensure that the memory
     * gets deallocated. Do NOT return directly from this function.
     */

    all = CONS(hp, make_small(0), NIL);
    hp += 2;

    while(is_list(env)) {
	Eterm tmp;
	Eterm* tp;

	tmp = CAR(list_val(env));
	if (is_not_tuple(tmp)) {
	    goto done;
	}
	tp = tuple_val(tmp);
	if (tp[0] != make_arityval(2)) {
	    goto done;
	}
	tmp = CONS(hp, make_small(0), NIL);
	hp += 2;
	if (tp[2] != am_false) {
	    tmp = CONS(hp, tp[2], tmp);
	    hp += 2;
	}
	tmp = CONS(hp, make_small('='), tmp);
	hp += 2;
	tmp = CONS(hp, tp[1], tmp);
	hp += 2;
	all = CONS(hp, tmp, all);
	hp += 2;
	env = CDR(list_val(env));
    }
    if (is_not_nil(env)) {
	goto done;
    }
    if ((n = io_list_len(all)) < 0) {
	goto done;
    }

    /*
     * Put the result in a binary (no risk for a memory leak that way).
     */
    (void) erts_new_heap_binary(p, NULL, n, &bytes);
    io_list_to_buf(all, (char*)bytes, n);

 done:
    erts_free(ERTS_ALC_T_TMP, temp_heap);
    return bytes;
}