Merge branch 'sverk/port_get_data-race-r16b03/OTP-12208' into maint-r16

* sverk/port_get_data-race-r16b03/OTP-12208:
  erts: Fix port data memory allocation bug
  erts: Mend port_set_data with non-immed data for halfword VM
  erts: Add test case for port_set_data and port_get_data
  erts: Fix race between port_set_data, port_get_data and port termination
  erts: Fix erlang:port_set_data/2 for non immediate data

3 files changed, 74 insertions, 7 deletions

diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 0bbd3e2dd5..6fdede1610 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -267,7 +267,6 @@ type	CODE_IX_LOCK_Q  SHORT_LIVED	SYSTEM		code_ix_lock_q
 type	PROC_INTERVAL	LONG_LIVED	SYSTEM		process_interval
 type    BUSY_CALLER_TAB	SHORT_LIVED	SYSTEM		busy_caller_table
 type    BUSY_CALLER	SHORT_LIVED	SYSTEM		busy_caller
-type	PORT_DATA_HEAP	STANDARD	SYSTEM		port_data_heap
 
 +if threads_no_smp
 # Need thread safe allocs, but std_alloc and fix_alloc are not;
@@ -361,6 +360,7 @@ type	NLINK_SH	STANDARD_LOW	PROCESSES	nlink_sh
 type	AINFO_REQ	STANDARD_LOW	SYSTEM		alloc_info_request
 type	SCHED_WTIME_REQ	STANDARD_LOW	SYSTEM		sched_wall_time_request
 type	GC_INFO_REQ	STANDARD_LOW	SYSTEM		gc_info_request
+type	PORT_DATA_HEAP	STANDARD_LOW	SYSTEM		port_data_heap
 
 +else # "fullword"
 
@@ -379,6 +379,7 @@ type	NLINK_SH	FIXED_SIZE	PROCESSES	nlink_sh
 type	AINFO_REQ	SHORT_LIVED	SYSTEM		alloc_info_request
 type	SCHED_WTIME_REQ	SHORT_LIVED	SYSTEM		sched_wall_time_request
 type	GC_INFO_REQ	SHORT_LIVED	SYSTEM		gc_info_request
+type	PORT_DATA_HEAP	STANDARD	SYSTEM		port_data_heap
 
 +endif
 
diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c
index 3cd53ef65d..ab473c6a08 100644
--- a/erts/emulator/beam/erl_bif_port.c
+++ b/erts/emulator/beam/erl_bif_port.c
@@ -472,7 +472,7 @@ cleanup_old_port_data(erts_aint_t data)
 	ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data;
 	size_t size;
 	ERTS_SMP_DATA_DEPENDENCY_READ_MEMORY_BARRIER;
-	size = sizeof(ErtsPortDataHeap) + pdhp->hsize*(sizeof(Eterm) - 1);
+	size = sizeof(ErtsPortDataHeap) + (pdhp->hsize-1)*sizeof(Eterm);
 	erts_schedule_thr_prgr_later_cleanup_op(free_port_data_heap,
 						(void *) pdhp,
 						&pdhp->later_op,
@@ -493,8 +493,8 @@ void
 erts_cleanup_port_data(Port *prt)
 {
     ASSERT(erts_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLGS_INVALID_LOOKUP);
-    cleanup_old_port_data(erts_smp_atomic_read_nob(&prt->data));
-    erts_smp_atomic_set_nob(&prt->data, (erts_aint_t) THE_NON_VALUE);
+    cleanup_old_port_data(erts_smp_atomic_xchg_nob(&prt->data,
+						   (erts_aint_t) NULL));
 }
 
 Uint
@@ -508,7 +508,7 @@ erts_port_data_size(Port *prt)
     }
     else {
 	ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data;
-	return (Uint) sizeof(ErtsPortDataHeap) + pdhp->hsize*(sizeof(Eterm)-1);
+	return (Uint) sizeof(ErtsPortDataHeap) + (pdhp->hsize-1)*sizeof(Eterm);
     }
 }
 
@@ -550,10 +550,11 @@ BIF_RETTYPE port_set_data_2(BIF_ALIST_2)
 
 	hsize = size_object(BIF_ARG_2);
 	pdhp = erts_alloc(ERTS_ALC_T_PORT_DATA_HEAP,
-			  sizeof(ErtsPortDataHeap) + hsize*(sizeof(Eterm)-1));
+			  sizeof(ErtsPortDataHeap) + (hsize-1)*sizeof(Eterm));
 	hp = &pdhp->heap[0];
 	pdhp->off_heap.first = NULL;
 	pdhp->off_heap.overhead = 0;
+	pdhp->hsize = hsize;
 	pdhp->data = copy_struct(BIF_ARG_2, hsize, &hp, &pdhp->off_heap);
 	data = (erts_aint_t) pdhp;
 	ASSERT((data & 0x3) == 0);
@@ -561,8 +562,14 @@ BIF_RETTYPE port_set_data_2(BIF_ALIST_2)
 
     data = erts_smp_atomic_xchg_wb(&prt->data, data);
 
+    if (data == (erts_aint_t)NULL) {
+	/* Port terminated by racing thread */
+	data = erts_smp_atomic_xchg_wb(&prt->data, data);
+	ASSERT(data != (erts_aint_t)NULL);
+	cleanup_old_port_data(data);
+	BIF_ERROR(BIF_P, BADARG);
+    }
     cleanup_old_port_data(data);
-
     BIF_RET(am_true);
 }
 
@@ -581,6 +588,8 @@ BIF_RETTYPE port_get_data_1(BIF_ALIST_1)
         BIF_ERROR(BIF_P, BADARG);
 
     data = erts_smp_atomic_read_ddrb(&prt->data);
+    if (data == (erts_aint_t)NULL)
+        BIF_ERROR(BIF_P, BADARG);  /* Port terminated by racing thread */
 
     if ((data & 0x3) != 0) {
 	res = (Eterm) (UWord) data;
diff --git a/erts/emulator/test/port_SUITE.erl b/erts/emulator/test/port_SUITE.erl
index fcd4457c34..8c904410b0 100644
--- a/erts/emulator/test/port_SUITE.erl
+++ b/erts/emulator/test/port_SUITE.erl
@@ -90,6 +90,7 @@
 	 mix_up_ports/1, otp_5112/1, otp_5119/1, otp_6224/1,
 	 exit_status_multi_scheduling_block/1, ports/1,
 	 spawn_driver/1, spawn_executable/1, close_deaf_port/1,
+	 port_setget_data/1,
 	 unregister_name/1, parallelism_option/1]).
 
 -export([do_iter_max_ports/2]).
@@ -115,6 +116,7 @@ all() ->
      mix_up_ports, otp_5112, otp_5119,
      exit_status_multi_scheduling_block, ports, spawn_driver,
      spawn_executable, close_deaf_port, unregister_name,
+     port_setget_data,
      parallelism_option].
 
 groups() -> 
@@ -2287,6 +2289,61 @@ close_deaf_port_1(N, Cmd) ->
 	    {comment, "Could not spawn more than " ++ integer_to_list(N) ++ " OS processes."}
     end.
 
+%% Test undocumented port_set_data/2 and port_get_data/1
+%% Hammer from multiple processes a while
+%% and then abrubtly close the port (OTP-12208).
+port_setget_data(Config) when is_list(Config) ->
+    ok = load_driver(?config(data_dir, Config), "echo_drv"),
+    Port = erlang:open_port({spawn_driver, "echo_drv"}, []),
+
+    NSched = erlang:system_info(schedulers_online),
+    HeapData = {1,2,3,<<"A heap binary">>,fun()->"This is fun"end,
+	       list_to_binary(lists:seq(1,100))},
+    PRs = lists:map(fun(I) ->
+			    spawn_opt(fun() -> port_setget_data_hammer(Port,HeapData,false,1) end,
+				      [monitor, {scheduler, I rem NSched}])
+		    end,
+		    lists:seq(1,10)),
+    receive after 100 -> ok end,
+    Papa = self(),
+    lists:foreach(fun({Pid,_}) -> Pid ! {Papa,prepare_for_close} end, PRs),
+    lists:foreach(fun({Pid,_}) ->
+			  receive {Pid,prepare_for_close} -> ok end
+		  end,
+		  PRs),
+    port_close(Port),
+    lists:foreach(fun({Pid,Ref}) ->
+			  receive {'DOWN', Ref, process, Pid, normal} -> ok end
+		  end,
+		  PRs),
+    ok.
+
+port_setget_data_hammer(Port, HeapData, IsSet0, N) ->
+    Rand = random:uniform(3),
+    IsSet1 = try case Rand of
+		     1 -> true = erlang:port_set_data(Port, atom), true;
+		     2 -> true = erlang:port_set_data(Port, HeapData), true;
+		     3 -> case erlang:port_get_data(Port) of
+			      atom -> true;
+			      HeapData -> true;
+			      undefined -> false=IsSet0
+			  end
+		 end
+    catch
+	error:badarg ->
+	    true = get(prepare_for_close),
+	    io:format("~p did ~p rounds before port closed\n", [self(), N]),
+	    exit(normal)
+    end,
+    receive {Papa, prepare_for_close} ->
+	    put(prepare_for_close, true),
+	    Papa ! {self(),prepare_for_close}
+    after 0 ->
+	    ok
+    end,
+    port_setget_data_hammer(Port, HeapData, IsSet1, N+1).
+
+
 wait_until(Fun) ->
     case catch Fun() of
 	true ->