Add support for busy port message queue

15 files changed, 779 insertions, 210 deletions

diff --git a/erts/doc/src/driver_entry.xml b/erts/doc/src/driver_entry.xml
index a2efdf3ebc..d65a38570d 100644
--- a/erts/doc/src/driver_entry.xml
+++ b/erts/doc/src/driver_entry.xml
@@ -377,11 +377,11 @@ typedef struct erl_drv_entry {
 
       <tag><marker id="driver_flags"/>int driver_flags</tag>
       <item>
-        <p>This field is used to pass driver capability information to the
-          runtime system. If the <c>extended_marker</c> field equals
-          <c>ERL_DRV_EXTENDED_MARKER</c>, it should contain <c>0</c> or
-          driver flags (<c>ERL_DRV_FLAG_*</c>) ored bitwise. Currently
-          the following driver flags exist:
+        <p>This field is used to pass driver capability and other
+	  information to the runtime system. If the
+	  <c>extended_marker</c> field equals <c>ERL_DRV_EXTENDED_MARKER</c>,
+	  it should contain <c>0</c> or  driver flags (<c>ERL_DRV_FLAG_*</c>)
+	  ored bitwise. Currently the following driver flags exist:
         </p>
         <taglist>
           <tag><c>ERL_DRV_FLAG_USE_PORT_LOCKING</c></tag>
@@ -404,6 +404,12 @@ typedef struct erl_drv_entry {
 	    by the Erlang distribution (the behaviour has always been
 	    required by drivers used by the distribution).
           </item>
+          <tag><c>ERL_DRV_FLAG_NO_BUSY_MSGQ</c></tag>
+          <item>Disable busy port message queue functionality. For
+	  more information, see the documentation of the
+	  <seealso marker="erl_driver#erl_drv_busy_msgq_limits">erl_drv_busy_msgq_limits()</seealso>
+	  function.
+          </item>
         </taglist> 
       </item>
       <tag>void *handle2</tag>
diff --git a/erts/doc/src/erl_driver.xml b/erts/doc/src/erl_driver.xml
index fcce962557..903b8dfe64 100644
--- a/erts/doc/src/erl_driver.xml
+++ b/erts/doc/src/erl_driver.xml
@@ -1500,16 +1500,81 @@ typedef struct ErlIOVec {
       </desc>
     </func>
     <func>
+      <name><ret>void</ret><nametext>erl_drv_busy_msgq_limits(ErlDrvPort port, ErlDrvSizeT *low, ErlDrvSizeT *high)</nametext></name>
+      <fsummary>Set and get limits for busy port message queue</fsummary>
+      <desc>
+        <marker id="erl_drv_busy_msgq_limits"></marker>
+        <p>Sets and gets limits that will be used for controling the
+	busy state of the port message queue.</p>
+	<p>The port message queue will be set into a busy
+	state when the amount of command data queued on the
+	message	queue reaches the <c>high</c> limit. The port
+	message	queue will be set into a not busy state when the
+	amount of command data queued on the message queue falls
+	below the <c>low</c> limit. Command data is in this
+	context data passed to the port using either
+	<c>Port ! {Owner, {command, Data}}</c>, or
+	<c>port_command/[2,3]</c>. Note that these limits
+	only concerns command data that have not yet reached the
+	port. The <seealso marker="#set_busy_port">busy port</seealso>
+	feature can be used for data that has reached the port.</p>
+
+	<p>Valid limits are values in the range
+	<c>[ERL_DRV_BUSY_MSGQ_LIM_MIN, ERL_DRV_BUSY_MSGQ_LIM_MAX]</c>.
+	Limits will be automatically adjusted to be sane. That is,
+	the system will	adjust values so that the low limit used is
+	lower or equal to the high limit used. By default the high
+	limit will be 8 kB and the low limit will be 4 kB.</p>
+
+	<p>By passing a pointer to an integer variable containing
+	the value <c>ERL_DRV_BUSY_MSGQ_READ_ONLY</c>, currently used
+	limit will be read and written back to the integer variable.
+	A new limit can be set by passing a pointer to an integer
+	variable containing a valid limit. The passed value will be
+	written to the internal limit. The internal limit will then
+	be adjusted. After this the adjusted limit will be written
+	back to the integer variable from which the new value was
+	read. Values are in bytes.</p>
+
+	<p>The busy message queue feature can be disabled either
+	by setting the <c>ERL_DRV_FLAG_NO_BUSY_MSGQ</c>
+	<seealso marker="driver_entry#driver_flags">driver flag</seealso>
+	in the <seealso marker="driver_entry">driver_entry</seealso>
+	used by the driver, or by calling this function with
+	<c>ERL_DRV_BUSY_MSGQ_DISABLED</c> as a limit (either low or
+	high). When this feature has been disabled it cannot be
+	enabled again. When reading the limits both of them
+	will be <c>ERL_DRV_BUSY_MSGQ_DISABLED</c>, if this
+	feature has been disabled.</p>
+
+	<p>Processes sending command data to the port will be suspended
+	if either the port is busy or if the port message queue is
+	busy. Suspended processes will be resumed when neither the
+	port is busy, nor the port message queue is busy.</p>
+
+	<p>For information about busy port functionality
+	see the documentation of the
+	<seealso marker="#set_busy_port">set_busy_port()</seealso>
+	function.</p>
+      </desc>
+    </func>
+    <func>
       <name><ret>void</ret><nametext>set_busy_port(ErlDrvPort port, int on)</nametext></name>
       <fsummary>Signal or unsignal port as busy</fsummary>
       <desc>
         <marker id="set_busy_port"></marker>
-        <p>This function set and resets the busy status of the port. If
-          <c>on</c> is 1, the port is set to busy, if it's 0 the port
-          is set to not busy.</p>
-        <p>When the port is busy, sending to it with <c>Port ! Data</c>
-          or <c>port_command/2</c>, will block the port owner process,
-          until the port is signaled as not busy.</p>
+        <p>This function set and unset the busy state of the port. If
+          <c>on</c> is non-zero, the port is set to busy, if it's zero the port
+          is set to not busy. You typically want to combine
+	  this feature with the <seealso marker="#erl_drv_busy_msgq_limits">busy
+	  port message queue</seealso> functionality.</p>
+	<p>Processes sending command data to the port will be suspended
+	if either the port is busy or if the port message queue
+	is busy. Suspended processes will be resumed when neither the
+	port is busy, nor the port message queue is busy. Command data
+	is in this context data passed to the port using either
+	<c>Port ! {Owner, {command, Data}}</c>, or
+	<c>port_command/[2,3]</c>.</p>
 	<p>If the
 	   <seealso marker="driver_entry#driver_flags"><![CDATA[ERL_DRV_FLAG_SOFT_BUSY]]></seealso>
 	   has been set in the
@@ -1518,6 +1583,10 @@ typedef struct ErlIOVec {
 	   <seealso marker="erlang#port_command/3">port_command(Port, Data, [force])</seealso>
 	   even though the driver has signaled that it is busy.
 	</p>
+	<p>For information about busy port message queue functionality
+	see the documentation of the
+	<seealso marker="#erl_drv_busy_msgq_limits">erl_drv_busy_msgq_limits()</seealso>
+	function.</p>
       </desc>
     </func>
     <func>
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index a939b90eac..fd5efd89f1 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -2039,7 +2039,7 @@ erts_dist_command(Port *prt, int reds_limit)
     if (reds > reds_limit)
 	goto preempted;
 
-    if (!(sched_flags & ERTS_PTS_FLG_BUSY) && foq.first) {
+    if (!(sched_flags & ERTS_PTS_FLG_BUSY_PORT) && foq.first) {
 	int preempt = 0;
 	do {
 	    Uint size;
@@ -2057,7 +2057,7 @@ erts_dist_command(Port *prt, int reds_limit)
 	    free_dist_obuf(fob);
 	    sched_flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
 	    preempt = reds > reds_limit || (sched_flags & ERTS_PTS_FLG_EXIT);
-	    if (sched_flags & ERTS_PTS_FLG_BUSY)
+	    if (sched_flags & ERTS_PTS_FLG_BUSY_PORT)
 		break;
 	} while (foq.first && !preempt);
 	if (!foq.first)
@@ -2066,7 +2066,7 @@ erts_dist_command(Port *prt, int reds_limit)
 	    goto preempted;
     }
 
-    if (sched_flags & ERTS_PTS_FLG_BUSY) {
+    if (sched_flags & ERTS_PTS_FLG_BUSY_PORT) {
 	if (oq.first) {
 	    ErtsDistOutputBuf *ob;
 	    int preempt;
@@ -2139,7 +2139,7 @@ erts_dist_command(Port *prt, int reds_limit)
 	    free_dist_obuf(fob);
 	    sched_flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
 	    preempt = reds > reds_limit || (sched_flags & ERTS_PTS_FLG_EXIT);
-	    if ((sched_flags & ERTS_PTS_FLG_BUSY) && oq.first && !preempt)
+	    if ((sched_flags & ERTS_PTS_FLG_BUSY_PORT) && oq.first && !preempt)
 		goto finalize_only;
 	}
 
@@ -2168,7 +2168,7 @@ erts_dist_command(Port *prt, int reds_limit)
 	ASSERT(dep->qsize >= obufsize);
 	dep->qsize -= obufsize;
 	obufsize = 0;
-	if (!(sched_flags & ERTS_PTS_FLG_BUSY)
+	if (!(sched_flags & ERTS_PTS_FLG_BUSY_PORT)
 	    && (dep->qflgs & ERTS_DE_QFLG_BUSY)
 	    && dep->qsize < erts_dist_buf_busy_limit) {
 	    ErtsProcList *suspendees;
@@ -2692,6 +2692,15 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3)
 
     erts_atomic32_read_bor_nob(&pp->state, ERTS_PORT_SFLG_DISTRIBUTION);
 
+    /*
+     * Dist-ports do not use the "busy port message queue" functionality, but
+     * instead use "busy dist entry" functionality.
+     */
+    {
+	ErlDrvSizeT disable = ERL_DRV_BUSY_MSGQ_DISABLED;
+	erl_drv_busy_msgq_limits((ErlDrvPort) pp, &disable, NULL);
+    }
+
     pp->dist_entry = dep;
 
     dep->version = version;
diff --git a/erts/emulator/beam/erl_driver.h b/erts/emulator/beam/erl_driver.h
index fb9e92e44b..cbc7bd4ede 100644
--- a/erts/emulator/beam/erl_driver.h
+++ b/erts/emulator/beam/erl_driver.h
@@ -157,6 +157,7 @@ typedef struct {
 
 #define ERL_DRV_FLAG_USE_PORT_LOCKING	(1 << 0)
 #define ERL_DRV_FLAG_SOFT_BUSY		(1 << 1)
+#define ERL_DRV_FLAG_NO_BUSY_MSGQ	(1 << 2)
 
 /*
  * Integer types
@@ -384,9 +385,18 @@ typedef struct erl_drv_entry {
     ErlDrvEntry* driver_init(void)
 #endif
 
+#define ERL_DRV_BUSY_MSGQ_DISABLED	(~((ErlDrvSizeT) 0))
+#define ERL_DRV_BUSY_MSGQ_READ_ONLY	((ErlDrvSizeT) 0)
+#define ERL_DRV_BUSY_MSGQ_LIM_MAX	(ERL_DRV_BUSY_MSGQ_DISABLED - 1)
+#define ERL_DRV_BUSY_MSGQ_LIM_MIN	((ErlDrvSizeT) 1)
+
 /*
  * These are the functions available for driver writers.
  */
+EXTERN void erl_drv_busy_msgq_limits(ErlDrvPort port,
+				     ErlDrvSizeT *low,
+				     ErlDrvSizeT *high);
+
 EXTERN int driver_select(ErlDrvPort port, ErlDrvEvent event, int mode, int on);
 EXTERN int driver_event(ErlDrvPort port, ErlDrvEvent event, 
 			ErlDrvEventData event_data);
diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h
index 1eb6dea710..67c8c28f3f 100644
--- a/erts/emulator/beam/erl_port.h
+++ b/erts/emulator/beam/erl_port.h
@@ -761,15 +761,36 @@ erts_port_driver_callback_epilogue(Port *prt, erts_aint32_t *statep)
 
 #endif /* #if ERTS_GLB_INLINE_INCL_FUNC_DEF */
 
+void erts_port_resume_procs(Port *);
+
 struct binary;
 
-#define ERTS_P2P_SIG_DATA_FLG_BANG_OP		(1 << 0)
-#define ERTS_P2P_SIG_DATA_FLG_REPLY		(1 << 1)
-#define ERTS_P2P_SIG_DATA_FLG_NOSUSPEND		(1 << 2)
-#define ERTS_P2P_SIG_DATA_FLG_FORCE		(1 << 3)
-#define ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT	(1 << 4)
-#define ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK	(1 << 5)
-#define ERTS_P2P_SIG_DATA_FLG_SCHED		(1 << 6)
+#define ERTS_P2P_SIG_TYPE_BAD			0
+#define ERTS_P2P_SIG_TYPE_OUTPUT		1
+#define ERTS_P2P_SIG_TYPE_OUTPUTV		2
+#define ERTS_P2P_SIG_TYPE_CONNECT		3
+#define ERTS_P2P_SIG_TYPE_EXIT			4
+#define ERTS_P2P_SIG_TYPE_CONTROL		5
+#define ERTS_P2P_SIG_TYPE_CALL			6
+#define ERTS_P2P_SIG_TYPE_INFO			7
+#define ERTS_P2P_SIG_TYPE_LINK			8
+#define ERTS_P2P_SIG_TYPE_UNLINK		9
+#define ERTS_P2P_SIG_TYPE_SET_DATA		10
+#define ERTS_P2P_SIG_TYPE_GET_DATA		11
+
+#define ERTS_P2P_SIG_TYPE_BITS			4
+#define ERTS_P2P_SIG_TYPE_MASK \
+    ((1 << ERTS_P2P_SIG_TYPE_BITS) - 1)
+
+#define ERTS_P2P_SIG_DATA_FLG(N) \
+    (1 << (ERTS_P2P_SIG_TYPE_BITS + (N)))
+#define ERTS_P2P_SIG_DATA_FLG_BANG_OP		ERTS_P2P_SIG_DATA_FLG(0)
+#define ERTS_P2P_SIG_DATA_FLG_REPLY		ERTS_P2P_SIG_DATA_FLG(1)
+#define ERTS_P2P_SIG_DATA_FLG_NOSUSPEND		ERTS_P2P_SIG_DATA_FLG(2)
+#define ERTS_P2P_SIG_DATA_FLG_FORCE		ERTS_P2P_SIG_DATA_FLG(3)
+#define ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT	ERTS_P2P_SIG_DATA_FLG(4)
+#define ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK	ERTS_P2P_SIG_DATA_FLG(5)
+#define ERTS_P2P_SIG_DATA_FLG_SCHED		ERTS_P2P_SIG_DATA_FLG(6)
 
 struct ErtsProc2PortSigData_ {
     int flags;
@@ -823,6 +844,35 @@ struct ErtsProc2PortSigData_ {
     } u;
 } ;
 
+ERTS_GLB_INLINE int
+erts_proc2port_sig_is_command_op(ErtsProc2PortSigData *sigdp);
+ERTS_GLB_INLINE ErlDrvSizeT
+erts_proc2port_sig_command_data_size(ErtsProc2PortSigData *sigdp);
+
+#if ERTS_GLB_INLINE_INCL_FUNC_DEF
+
+ERTS_GLB_INLINE int
+erts_proc2port_sig_is_command_op(ErtsProc2PortSigData *sigdp)
+{
+    switch (sigdp->flags & ERTS_P2P_SIG_TYPE_MASK) {
+    case ERTS_P2P_SIG_TYPE_OUTPUT:	return !0;
+    case ERTS_P2P_SIG_TYPE_OUTPUTV:	return !0;
+    default:				return 0;
+    }
+}
+
+ERTS_GLB_INLINE ErlDrvSizeT
+erts_proc2port_sig_command_data_size(ErtsProc2PortSigData *sigdp)
+{
+    switch (sigdp->flags & ERTS_P2P_SIG_TYPE_MASK) {
+    case ERTS_P2P_SIG_TYPE_OUTPUT:	return sigdp->u.output.size;
+    case ERTS_P2P_SIG_TYPE_OUTPUTV:	return sigdp->u.outputv.evp->size;
+    default:				return (ErlDrvSizeT) 0;
+    }
+}
+
+#endif
+
 #define ERTS_PROC2PORT_SIG_EXEC			0
 #define ERTS_PROC2PORT_SIG_ABORT		1
 #define ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND	2
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index b6b113b753..89f6a4e774 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -539,103 +539,257 @@ set_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp)
     }
 }
 
+
 /*
- * No-suspend handles.
+ * Busy port queue management
  */
 
-#ifdef ERTS_SMP
-static void
-free_port_task_handle_list(void *vpthlp)
+static erts_aint32_t
+check_unset_busy_port_q(Port *pp,
+			erts_aint32_t flags,
+			ErtsPortTaskBusyPortQ *bpq)
 {
-    erts_free(ERTS_ALC_T_PT_HNDL_LIST, vpthlp);
-}
-#endif
+    ErlDrvSizeT qsize, low;
+    int resume_procs = 0;
 
-static void
-schedule_port_task_handle_list_free(ErtsPortTaskHandleList *pthlp)
-{
-#ifdef ERTS_SMP
-    erts_schedule_thr_prgr_later_op(free_port_task_handle_list,
-				    (void *) pthlp,
-				    &pthlp->u.release);
-#else
-    erts_free(ERTS_ALC_T_PT_HNDL_LIST, pthlp);
-#endif
+    ASSERT(bpq);
+    ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp));
+
+    erts_port_task_sched_lock(&pp->sched);
+    qsize = (ErlDrvSizeT) erts_smp_atomic_read_nob(&bpq->size);
+    low = (ErlDrvSizeT) erts_smp_atomic_read_nob(&bpq->low);
+    if (qsize < low) {
+	erts_aint32_t mask = ~(ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q
+			       | ERTS_PTS_FLG_BUSY_PORT_Q);
+	flags = erts_smp_atomic32_read_band_relb(&pp->sched.flags, mask);
+	if ((flags & ERTS_PTS_FLGS_BUSY) == ERTS_PTS_FLG_BUSY_PORT_Q)
+	    resume_procs = 1;
+    }
+    else if (flags & ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q) {
+	flags = erts_smp_atomic32_read_band_relb(&pp->sched.flags,
+						 ~ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q);
+	flags &= ~ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q;
+    }
+    erts_port_task_sched_unlock(&pp->sched);
+    if (resume_procs)
+	erts_port_resume_procs(pp);
+
+    return flags;
 }
 
 static ERTS_INLINE void
-abort_nosuspend_task(ErtsPortTaskType type,
-		     ErtsPortTaskTypeData *tdp)
+aborted_proc2port_data(Port *pp, ErlDrvSizeT size)
 {
+    ErtsPortTaskBusyPortQ *bpq;
+    erts_aint32_t flags;
+    ErlDrvSizeT qsz;
 
-    if (type != ERTS_PORT_TASK_PROC_SIG)
-	ERTS_INTERNAL_ERROR("Invalid no-suspend port task type");
+    ASSERT(pp->sched.taskq.bpq);
 
-    tdp->psig.callback(NULL,
-		       ERTS_PORT_SFLG_INVALID,
-		       ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND,
-		       &tdp->psig.data);
-}
+    if (size == 0)
+	return;
 
+    bpq = pp->sched.taskq.bpq;
 
-static void
-save_nosuspend_handle(Port *pp, ErtsPortTask *ptp)
-{
-    erts_aint32_t act;
-    ErtsPortTaskHandleList *pthlp = erts_alloc(ERTS_ALC_T_PT_HNDL_LIST,
-					       sizeof(ErtsPortTaskHandleList));
+    qsz = (ErlDrvSizeT) erts_smp_atomic_add_read_acqb(&bpq->size,
+						      (erts_aint_t) -size);
+    ASSERT(qsz + size > qsz);
+    flags = erts_smp_atomic32_read_nob(&pp->sched.flags);
+    ASSERT(pp->sched.taskq.bpq);
+    if ((flags & (ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q
+		  | ERTS_PTS_FLG_BUSY_PORT_Q)) != ERTS_PTS_FLG_BUSY_PORT_Q)
+	return;
+    if (qsz < (ErlDrvSizeT) erts_smp_atomic_read_nob(&bpq->low))
+	erts_smp_atomic32_read_bor_nob(&pp->sched.flags,
+				       ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q);
+}
 
-    set_handle(ptp, &pthlp->handle);
+static ERTS_INLINE void
+dequeued_proc2port_data(Port *pp, ErlDrvSizeT size)
+{
+    ErtsPortTaskBusyPortQ *bpq;
+    erts_aint32_t flags;
+    ErlDrvSizeT qsz;
 
-    ASSERT(ptp == handle2task(&pthlp->handle));
-    ASSERT(ptp->u.alive.handle == &pthlp->handle);
+    ASSERT(pp->sched.taskq.bpq);
 
-    act = erts_smp_atomic32_read_nob(&pp->sched.flags);
+    if (size == 0)
+	return;
 
-    if (!(act & ERTS_PTS_FLG_BUSY)) {
+    bpq = pp->sched.taskq.bpq;
 
-	erts_port_task_sched_lock(&pp->sched);
-	pthlp->u.next = pp->sched.taskq.local.busy.nosuspend;
-	pp->sched.taskq.local.busy.nosuspend = pthlp;
-	erts_port_task_sched_unlock(&pp->sched);
+    qsz = (ErlDrvSizeT) erts_smp_atomic_add_read_acqb(&bpq->size,
+						      (erts_aint_t) -size);
+    ASSERT(qsz + size > qsz);
+    flags = erts_smp_atomic32_read_nob(&pp->sched.flags);
+    if (!(flags & ERTS_PTS_FLG_BUSY_PORT_Q))
+	return;
+    if (qsz < (ErlDrvSizeT) erts_smp_atomic_read_acqb(&bpq->low))
+	check_unset_busy_port_q(pp, flags, bpq);
+}
 
-	act = erts_smp_atomic32_read_nob(&pp->sched.flags);
+static ERTS_INLINE erts_aint32_t
+enqueue_proc2port_data(Port *pp,
+		       ErtsProc2PortSigData *sigdp,
+		       erts_aint32_t flags)
+{
+    ErtsPortTaskBusyPortQ *bpq = pp->sched.taskq.bpq;
+    if (sigdp && bpq) {
+	ErlDrvSizeT size = erts_proc2port_sig_command_data_size(sigdp);
+	if (size) {
+	    erts_aint_t asize = erts_smp_atomic_add_read_acqb(&bpq->size,
+							      (erts_aint_t) size);
+	    ErlDrvSizeT qsz = (ErlDrvSizeT) asize;
+
+	    ASSERT(qsz - size < qsz);
+
+	    if (!(flags & ERTS_PTS_FLG_BUSY_PORT_Q) && qsz > bpq->high) {
+		flags = erts_smp_atomic32_read_bor_acqb(&pp->sched.flags,
+							ERTS_PTS_FLG_BUSY_PORT_Q);
+		flags |= ERTS_PTS_FLG_BUSY_PORT_Q;
+		qsz = (ErlDrvSizeT) erts_smp_atomic_read_acqb(&bpq->size);
+		if (qsz < (ErlDrvSizeT) erts_smp_atomic_read_nob(&bpq->low)) {
+		    flags = (erts_smp_atomic32_read_bor_relb(
+				 &pp->sched.flags,
+				 ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q));
+		    flags |= ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q;
+		}
+	    }
+	    ASSERT(!(flags & ERTS_PTS_FLG_EXIT));
+	}
+    }
+    return flags;
+}
 
-	while (1) {
-	    erts_aint32_t exp, new;
+/*
+ * erl_drv_busy_msgq_limits() is called by drivers either reading or
+ * writing the limits.
+ *
+ * A limit of zero is interpreted as a read only request (using a
+ * limit of zero would not be useful). Other values are interpreted
+ * as a write-read request.
+ */
 
-	    if ((act & (ERTS_PTS_FLG_BUSY|ERTS_PTS_FLG_HAVE_NS_TASKS))
-		== ERTS_PTS_FLG_HAVE_NS_TASKS)
-		return;
+void
+erl_drv_busy_msgq_limits(ErlDrvPort dport, ErlDrvSizeT *lowp, ErlDrvSizeT *highp)
+{
+    Port *pp = erts_drvport2port(dport, NULL);
+    ErtsPortTaskBusyPortQ *bpq = pp->sched.taskq.bpq;
+    int written = 0, resume_procs = 0;
+    ErlDrvSizeT low, high;
+
+    if (!pp || !bpq) {
+	if (lowp)
+	    *lowp = ERL_DRV_BUSY_MSGQ_DISABLED;
+	if (highp)
+	    *highp = ERL_DRV_BUSY_MSGQ_DISABLED;
+	return;
+    }
 
-	    if (act & ERTS_PTS_FLG_BUSY) {
-		erts_aint32_t s;
-		s = erts_smp_atomic32_cmpxchg_nob(&ptp->state,
-						  ERTS_PT_STATE_ABORTED,
-						  ERTS_PT_STATE_SCHEDULED);
-		if (s == ERTS_PT_STATE_SCHEDULED) {
-		    reset_port_task_handle(&pthlp->handle);
-		    break; /* Abort task */
-		}
-		/* Else: someone else handled it */
-		return;
-	    }
+    low = lowp ? *lowp : 0;
+    high = highp ? *highp : 0;
 
-	    new = exp = act;
+    erts_port_task_sched_lock(&pp->sched);
 
-	    new |= ERTS_PTS_FLG_HAVE_NS_TASKS;
+    if (low == ERL_DRV_BUSY_MSGQ_DISABLED
+	|| high == ERL_DRV_BUSY_MSGQ_DISABLED) {
+	/* Disable busy msgq feature */
+	erts_aint32_t flags;
+	pp->sched.taskq.bpq = NULL;
+	flags = ~(ERTS_PTS_FLG_BUSY_PORT_Q|ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q);
+	flags = erts_smp_atomic32_read_band_acqb(&pp->sched.flags, flags);
+	if ((flags & ERTS_PTS_FLGS_BUSY) == ERTS_PTS_FLG_BUSY_PORT_Q)
+	    resume_procs = 1;
+    }
+    else {
 
-	    act = erts_smp_atomic32_cmpxchg_relb(&pp->sched.flags, new, exp);
-	    if (act == exp)
-		return;
+	if (!low)
+	    low = (ErlDrvSizeT) erts_smp_atomic_read_nob(&bpq->low);
+	else {
+	    if (bpq->high < low)
+		bpq->high = low;
+	    erts_smp_atomic_set_relb(&bpq->low, (erts_aint_t) low);
+	    written = 1;
+	}
+    
+	if (!high)
+	    high = bpq->high;
+	else {
+	    if (low > high) {
+		low = high;
+		erts_smp_atomic_set_relb(&bpq->low, (erts_aint_t) low);
+	    }
+	    bpq->high = high;
+	    written = 1;
+	}
 
+	if (written) {
+	    ErlDrvSizeT size = (ErlDrvSizeT) erts_smp_atomic_read_nob(&bpq->size);
+	    if (size > high)
+		erts_smp_atomic32_read_bor_relb(&pp->sched.flags,
+						ERTS_PTS_FLG_BUSY_PORT_Q);
+	    else if (size < low)
+		erts_smp_atomic32_read_bor_relb(&pp->sched.flags,
+						ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q);
 	}
     }
 
+    erts_port_task_sched_unlock(&pp->sched);
 
-    abort_nosuspend_task(ptp->type, &ptp->u.alive.td);
+    if (resume_procs)
+	erts_port_resume_procs(pp);
+    if (lowp)
+	*lowp = low;
+    if (highp)
+	*highp = high;
 }
 
+/*
+ * No-suspend handles.
+ */
+
+#ifdef ERTS_SMP
+static void
+free_port_task_handle_list(void *vpthlp)
+{
+    erts_free(ERTS_ALC_T_PT_HNDL_LIST, vpthlp);
+}
+#endif
+
+static void
+schedule_port_task_handle_list_free(ErtsPortTaskHandleList *pthlp)
+{
+#ifdef ERTS_SMP
+    erts_schedule_thr_prgr_later_op(free_port_task_handle_list,
+				    (void *) pthlp,
+				    &pthlp->u.release);
+#else
+    erts_free(ERTS_ALC_T_PT_HNDL_LIST, pthlp);
+#endif
+}
+
+static ERTS_INLINE void
+abort_nosuspend_task(Port *pp,
+		     ErtsPortTaskType type,
+		     ErtsPortTaskTypeData *tdp)
+{
+
+    ASSERT(type == ERTS_PORT_TASK_PROC_SIG);
+
+    if (!pp->sched.taskq.bpq)
+	tdp->psig.callback(NULL,
+			   ERTS_PORT_SFLG_INVALID,
+			   ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND,
+			   &tdp->psig.data);
+    else {
+	ErlDrvSizeT size = erts_proc2port_sig_command_data_size(&tdp->psig.data);
+	tdp->psig.callback(NULL,
+			   ERTS_PORT_SFLG_INVALID,
+			   ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND,
+			   &tdp->psig.data);
+	aborted_proc2port_data(pp, size);
+    }
+}
 
 static ErtsPortTaskHandleList *
 get_free_nosuspend_handles(Port *pp)
@@ -724,14 +878,29 @@ pop_port(ErtsRunQueue *runq)
  * Task queue operations
  */
 
-static ERTS_INLINE erts_aint32_t
-enqueue_task(Port *pp, ErtsPortTask *ptp)
+static ERTS_INLINE int
+enqueue_task(Port *pp,
+	     ErtsPortTask *ptp,
+	     ErtsProc2PortSigData *sigdp,
+	     ErtsPortTaskHandleList *ns_pthlp,
+	     erts_aint32_t *flagsp)
+
 {
+    int res;
+    erts_aint32_t fail_flags = ERTS_PTS_FLG_EXIT;
     erts_aint32_t flags;
     ptp->u.alive.next = NULL;
+    if (ns_pthlp)
+	fail_flags |= ERTS_PTS_FLG_BUSY_PORT;
     erts_port_task_sched_lock(&pp->sched);
     flags = erts_smp_atomic32_read_nob(&pp->sched.flags);
-    if (!(flags & ERTS_PTS_FLG_EXIT)) {
+    if (flags & fail_flags)
+	res = 0;
+    else {
+	if (ns_pthlp) {
+	    ns_pthlp->u.next = pp->sched.taskq.local.busy.nosuspend;
+	    pp->sched.taskq.local.busy.nosuspend = ns_pthlp;
+	}
 	if (pp->sched.taskq.in.last) {
 	    ASSERT(pp->sched.taskq.in.first);
 	    ASSERT(!pp->sched.taskq.in.last->u.alive.next);
@@ -744,9 +913,12 @@ enqueue_task(Port *pp, ErtsPortTask *ptp)
 	    pp->sched.taskq.in.first = ptp;
 	}
 	pp->sched.taskq.in.last = ptp;
+	flags = enqueue_proc2port_data(pp, sigdp, flags);
+	res = 1;
     }
     erts_port_task_sched_unlock(&pp->sched);
-    return flags;
+    *flagsp = flags;
+    return res;
 }
 
 static ERTS_INLINE void
@@ -754,7 +926,7 @@ prepare_exec(Port *pp, ErtsPortTask **execqp, int *processing_busy_q_p)
 {
     erts_aint32_t act = erts_smp_atomic32_read_nob(&pp->sched.flags);
 
-    if (!pp->sched.taskq.local.busy.first || (act & ERTS_PTS_FLG_BUSY)) {
+    if (!pp->sched.taskq.local.busy.first || (act & ERTS_PTS_FLG_BUSY_PORT)) {
 	*execqp = pp->sched.taskq.local.first;
 	*processing_busy_q_p = 0;
     }
@@ -799,10 +971,9 @@ finalize_exec(Port *pp, ErtsPortTask **execq, int processing_busy_q)
 
     *execq = NULL;
 
-    /* guess a likely value */
-    act = ERTS_PTS_FLG_EXEC;
-    if (execq)
-	act |= ERTS_PTS_FLG_HAVE_TASKS;
+    act = erts_smp_atomic32_read_nob(&pp->sched.flags);
+    if (act & ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q)
+	act = check_unset_busy_port_q(pp, act, pp->sched.taskq.bpq);
 
     while (1) {
 	erts_aint32_t new, exp;
@@ -829,9 +1000,12 @@ select_queue_for_exec(Port *pp, ErtsPortTask **execqp, int *processing_busy_q_p)
 {
     erts_aint32_t flags = erts_smp_atomic32_read_nob(&pp->sched.flags);
 
+    if (flags & ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q)
+	flags = check_unset_busy_port_q(pp, flags, pp->sched.taskq.bpq);
+
     ERTS_PT_DBG_CHK_TASK_QS(pp, *execqp, *processing_busy_q_p);
 
-    if (flags & ERTS_PTS_FLG_BUSY) {
+    if (flags & ERTS_PTS_FLG_BUSY_PORT) {
 	if (*processing_busy_q_p) {
 	    ErtsPortTask *ptp;
 
@@ -880,7 +1054,7 @@ check_task_for_exec(Port *pp,
 
 	ERTS_PT_DBG_CHK_TASK_QS(pp, ptp, *processing_busy_q_p);
 
-	if ((flags & ERTS_PTS_FLG_BUSY)
+	if ((flags & ERTS_PTS_FLG_BUSY_PORT)
 	    && (ptp->u.alive.flags & ERTS_PT_FLG_WAIT_BUSY)) {
 
 	    busy_wait_move_to_busy_queue(pp, ptp);
@@ -902,7 +1076,7 @@ check_task_for_exec(Port *pp,
     else {
 	/* Processing busy queue */
 
-	ASSERT(!(flags & ERTS_PTS_FLG_BUSY));
+	ASSERT(!(flags & ERTS_PTS_FLG_BUSY_PORT));
 
 	ERTS_PT_DBG_CHK_TASK_QS(pp, ptp, *processing_busy_q_p);
 
@@ -1019,10 +1193,7 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp)
 		erts_smp_atomic_dec_relb(&erts_port_task_outstanding_io_tasks);
 		break;
 	    case ERTS_PORT_TASK_PROC_SIG:
-		ptp->u.alive.td.psig.callback(NULL,
-					      ERTS_PORT_SFLG_INVALID,
-					      ERTS_PROC2PORT_SIG_ABORT,
-					      &ptp->u.alive.td.psig.data);
+		ERTS_INTERNAL_ERROR("Aborted process to port signal");
 		break;
 	    default:
 		break;
@@ -1042,14 +1213,15 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp)
 void
 erts_port_task_abort_nosuspend_tasks(Port *pp)
 {
+    erts_aint32_t flags;
     ErtsPortTaskHandleList *abort_list;
 #ifdef ERTS_SMP
     ErtsThrPrgrDelayHandle dhndl = ERTS_THR_PRGR_DHANDLE_INVALID;
 #endif
 
     erts_port_task_sched_lock(&pp->sched);
-    erts_smp_atomic32_read_band_nob(&pp->sched.flags,
-				    ~ERTS_PTS_FLG_HAVE_NS_TASKS);
+    flags = erts_smp_atomic32_read_band_nob(&pp->sched.flags,
+					    ~ERTS_PTS_FLG_HAVE_NS_TASKS);
     abort_list = pp->sched.taskq.local.busy.nosuspend;
     pp->sched.taskq.local.busy.nosuspend = NULL;
     erts_port_task_sched_unlock(&pp->sched);
@@ -1117,7 +1289,7 @@ erts_port_task_abort_nosuspend_tasks(Port *pp)
 #endif
 	schedule_port_task_handle_list_free(pthlp);
 
-	abort_nosuspend_task(type, &td);
+	abort_nosuspend_task(pp, type, &td);
     }
 }
 
@@ -1131,6 +1303,8 @@ erts_port_task_schedule(Eterm id,
 			ErtsPortTaskType type,
 			...)
 {
+    ErtsProc2PortSigData *sigdp = NULL;
+    ErtsPortTaskHandleList *ns_pthlp = NULL;
 #ifdef ERTS_SMP
     ErtsRunQueue *xrunq;
     ErtsThrPrgrDelayHandle dhndl;
@@ -1138,7 +1312,7 @@ erts_port_task_schedule(Eterm id,
     ErtsRunQueue *runq;
     Port *pp;
     ErtsPortTask *ptp = NULL;
-    erts_aint32_t act;
+    erts_aint32_t act, add_flags;
 
     if (pthp && erts_port_task_is_scheduled(pthp)) {
 	ASSERT(0);
@@ -1195,7 +1369,6 @@ erts_port_task_schedule(Eterm id,
 	break;
     }
     case ERTS_PORT_TASK_PROC_SIG: {
-	ErtsProc2PortSigData *sigdp;
 	va_list argp;
 	ASSERT(!pthp);
 	va_start(argp, type);
@@ -1204,31 +1377,40 @@ erts_port_task_schedule(Eterm id,
 	ptp->u.alive.td.psig.callback = va_arg(argp, ErtsProc2PortSigCallback);
  	ptp->u.alive.flags |= va_arg(argp, int);
 	va_end(argp);
-	if (ptp->u.alive.flags & ERTS_PT_FLG_NOSUSPEND)
-	    save_nosuspend_handle(pp, ptp);
-	else
+	if (!(ptp->u.alive.flags & ERTS_PT_FLG_NOSUSPEND))
 	    set_handle(ptp, pthp);
+	else {
+	    ns_pthlp = erts_alloc(ERTS_ALC_T_PT_HNDL_LIST,
+				  sizeof(ErtsPortTaskHandleList));
+	    set_handle(ptp, &ns_pthlp->handle);
+	}
 	break;
     }
     default:
 	break;
     }
 
-    act = enqueue_task(pp, ptp);
-    if (act & ERTS_PTS_FLG_EXIT) {
+    if (!enqueue_task(pp, ptp, sigdp, ns_pthlp, &act)) {
 	reset_handle(ptp);
-	goto fail;
+	if (ns_pthlp && !(act & ERTS_PTS_FLG_EXIT))
+	    goto abort_nosuspend;
+	else
+	    goto fail;
     }
 
+    add_flags = ERTS_PTS_FLG_HAVE_TASKS;
+    if (ns_pthlp)
+	add_flags |= ERTS_PTS_FLG_HAVE_NS_TASKS;
+
     while (1) {
 	erts_aint32_t new, exp;
 
-	if ((act & ERTS_PTS_FLG_HAVE_TASKS)
+	if ((act & add_flags) == add_flags
 	    && (act & (ERTS_PTS_FLG_IN_RUNQ|ERTS_PTS_FLG_EXEC)))
 	    goto done; /* Done */
 
 	new = exp = act;
-	new |= ERTS_PTS_FLG_HAVE_TASKS;
+	new |= add_flags;
 	if (!(act & (ERTS_PTS_FLG_IN_RUNQ|ERTS_PTS_FLG_EXEC)))
 	    new |= ERTS_PTS_FLG_IN_RUNQ;
 
@@ -1281,6 +1463,22 @@ done:
 
     return 0;
 
+abort_nosuspend:
+
+#ifdef ERTS_SMP
+    if (dhndl != ERTS_THR_PRGR_DHANDLE_MANAGED)
+	erts_port_dec_refc(pp);
+#endif
+
+    abort_nosuspend_task(pp, ptp->type, &ptp->u.alive.td);
+
+    ASSERT(ns_pthlp);
+    erts_free(ERTS_ALC_T_PT_HNDL_LIST, ns_pthlp);
+    if (ptp)
+	port_task_free(ptp);
+
+    return 0;
+
 fail:
 
 #ifdef ERTS_SMP
@@ -1288,6 +1486,9 @@ fail:
 	erts_port_dec_refc(pp);
 #endif
 
+    if (ns_pthlp)
+	erts_free(ERTS_ALC_T_PT_HNDL_LIST, ns_pthlp);
+
     if (ptp)
 	port_task_free(ptp);
 
@@ -1444,13 +1645,24 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 				  ptp->u.alive.td.io.event_data);
 	    io_tasks_executed++;
 	    break;
-	case ERTS_PORT_TASK_PROC_SIG:
+	case ERTS_PORT_TASK_PROC_SIG: {
+	    ErtsProc2PortSigData *sigdp = &ptp->u.alive.td.psig.data;
 	    ASSERT((state & ERTS_PORT_SFLGS_DEAD) == 0);
-	    reds += ptp->u.alive.td.psig.callback(pp,
-						  state,
-						  ERTS_PROC2PORT_SIG_EXEC,
-						  &ptp->u.alive.td.psig.data);
+	    if (!pp->sched.taskq.bpq)
+		reds += ptp->u.alive.td.psig.callback(pp,
+						      state,
+						      ERTS_PROC2PORT_SIG_EXEC,
+						      sigdp);
+	    else {
+		ErlDrvSizeT size = erts_proc2port_sig_command_data_size(sigdp);
+		reds += ptp->u.alive.td.psig.callback(pp,
+						      state,
+						      ERTS_PROC2PORT_SIG_EXEC,
+						      sigdp);
+		dequeued_proc2port_data(pp, size);
+	    }
 	    break;
+	}
 	case ERTS_PORT_TASK_DIST_CMD:
 	    reds += erts_dist_command(pp, CONTEXT_REDS-reds);
 	    break;
@@ -1633,12 +1845,23 @@ begin_port_cleanup(Port *pp, ErtsPortTask **execqp)
 		break;
 	    case ERTS_PORT_TASK_DIST_CMD:
 		break;
-	    case ERTS_PORT_TASK_PROC_SIG:
-		ptp->u.alive.td.psig.callback(NULL,
-					      ERTS_PORT_SFLG_INVALID,
-					      ERTS_PROC2PORT_SIG_ABORT_CLOSED,
-					      &ptp->u.alive.td.psig.data);
+	    case ERTS_PORT_TASK_PROC_SIG: {
+		ErtsProc2PortSigData *sigdp = &ptp->u.alive.td.psig.data;
+		if (!pp->sched.taskq.bpq)
+		    ptp->u.alive.td.psig.callback(NULL,
+						  ERTS_PORT_SFLG_INVALID,
+						  ERTS_PROC2PORT_SIG_ABORT_CLOSED,
+						  sigdp);
+		else {
+		    ErlDrvSizeT size = erts_proc2port_sig_command_data_size(sigdp);
+		    ptp->u.alive.td.psig.callback(NULL,
+						  ERTS_PORT_SFLG_INVALID,
+						  ERTS_PROC2PORT_SIG_ABORT_CLOSED,
+						  sigdp);
+		    aborted_proc2port_data(pp, size);
+		}
 		break;
+	    }
 	    default:
 		erl_exit(ERTS_ABORT_EXIT,
 			 "Invalid port task type: %d\n",
diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h
index c41f8104c7..9976604042 100644
--- a/erts/emulator/beam/erl_port_task.h
+++ b/erts/emulator/beam/erl_port_task.h
@@ -66,16 +66,20 @@ typedef enum {
 extern erts_smp_atomic_t erts_port_task_outstanding_io_tasks;
 #endif
 
-#define ERTS_PTS_FLG_IN_RUNQ		(((erts_aint32_t) 1) << 0)
-#define ERTS_PTS_FLG_EXEC		(((erts_aint32_t) 1) << 1)
-#define ERTS_PTS_FLG_HAVE_TASKS		(((erts_aint32_t) 1) << 2)
-#define ERTS_PTS_FLG_EXIT		(((erts_aint32_t) 1) << 3)
-#define ERTS_PTS_FLG_BUSY		(((erts_aint32_t) 1) << 4)
-#define ERTS_PTS_FLG_HAVE_BUSY_TASKS	(((erts_aint32_t) 1) << 5)
-#define ERTS_PTS_FLG_HAVE_NS_TASKS	(((erts_aint32_t) 1) << 6)
-#define ERTS_PTS_FLG_PARALLELISM	(((erts_aint32_t) 1) << 7)
-#define ERTS_PTS_FLG_FORCE_SCHED	(((erts_aint32_t) 1) << 8)
+#define ERTS_PTS_FLG_IN_RUNQ			(((erts_aint32_t) 1) <<  0)
+#define ERTS_PTS_FLG_EXEC			(((erts_aint32_t) 1) <<  1)
+#define ERTS_PTS_FLG_HAVE_TASKS			(((erts_aint32_t) 1) <<  2)
+#define ERTS_PTS_FLG_EXIT			(((erts_aint32_t) 1) <<  3)
+#define ERTS_PTS_FLG_BUSY_PORT			(((erts_aint32_t) 1) <<  4)
+#define ERTS_PTS_FLG_BUSY_PORT_Q		(((erts_aint32_t) 1) <<  5)
+#define ERTS_PTS_FLG_CHK_UNSET_BUSY_PORT_Q	(((erts_aint32_t) 1) <<  6)
+#define ERTS_PTS_FLG_HAVE_BUSY_TASKS		(((erts_aint32_t) 1) <<  7)
+#define ERTS_PTS_FLG_HAVE_NS_TASKS		(((erts_aint32_t) 1) <<  8)
+#define ERTS_PTS_FLG_PARALLELISM		(((erts_aint32_t) 1) <<  9)
+#define ERTS_PTS_FLG_FORCE_SCHED		(((erts_aint32_t) 1) << 10)
 
+#define ERTS_PTS_FLGS_BUSY \
+    (ERTS_PTS_FLG_BUSY_PORT | ERTS_PTS_FLG_BUSY_PORT_Q)
 
 #define ERTS_PTS_FLGS_FORCE_SCHEDULE_OP		\
     (ERTS_PTS_FLG_EXIT				\
@@ -84,6 +88,15 @@ extern erts_smp_atomic_t erts_port_task_outstanding_io_tasks;
      | ERTS_PTS_FLG_EXEC			\
      | ERTS_PTS_FLG_FORCE_SCHED)
 
+#define ERTS_PORT_TASK_DEFAULT_BUSY_PORT_Q_HIGH			8192
+#define ERTS_PORT_TASK_DEFAULT_BUSY_PORT_Q_LOW			4096
+
+typedef struct {
+    ErlDrvSizeT high;
+    erts_smp_atomic_t low;
+    erts_smp_atomic_t size;
+}  ErtsPortTaskBusyPortQ;
+
 typedef struct ErtsPortTask_ ErtsPortTask;
 typedef struct ErtsPortTaskBusyCallerTable_ ErtsPortTaskBusyCallerTable;
 typedef struct ErtsPortTaskHandleList_ ErtsPortTaskHandleList;
@@ -104,6 +117,7 @@ typedef struct {
 	    ErtsPortTask *first;
 	    ErtsPortTask *last;
 	} in;
+	ErtsPortTaskBusyPortQ *bpq;
     } taskq;
     erts_smp_atomic32_t flags;
 #ifdef ERTS_SMP
@@ -113,6 +127,8 @@ typedef struct {
 
 ERTS_GLB_INLINE void erts_port_task_handle_init(ErtsPortTaskHandle *pthp);
 ERTS_GLB_INLINE int erts_port_task_is_scheduled(ErtsPortTaskHandle *pthp);
+ERTS_GLB_INLINE void erts_port_task_pre_init_sched(ErtsPortTaskSched *ptsp,
+						   ErtsPortTaskBusyPortQ *bpq);
 ERTS_GLB_INLINE void erts_port_task_init_sched(ErtsPortTaskSched *ptsp,
 					       Eterm id);
 ERTS_GLB_INLINE void erts_port_task_fini_sched(ErtsPortTaskSched *ptsp);
@@ -138,6 +154,18 @@ erts_port_task_is_scheduled(ErtsPortTaskHandle *pthp)
     return ((void *) erts_smp_atomic_read_nob(pthp)) != NULL;
 }
 
+ERTS_GLB_INLINE void erts_port_task_pre_init_sched(ErtsPortTaskSched *ptsp,
+						   ErtsPortTaskBusyPortQ *bpq)
+{
+    if (bpq) {
+	erts_aint_t low = (erts_aint_t) ERTS_PORT_TASK_DEFAULT_BUSY_PORT_Q_LOW;
+	erts_smp_atomic_init_nob(&bpq->low, low);
+ 	bpq->high = (ErlDrvSizeT) ERTS_PORT_TASK_DEFAULT_BUSY_PORT_Q_HIGH;
+	erts_smp_atomic_init_nob(&bpq->size, (erts_aint_t) 0);
+    }
+    ptsp->taskq.bpq = bpq;
+}
+
 ERTS_GLB_INLINE void
 erts_port_task_init_sched(ErtsPortTaskSched *ptsp, Eterm instr_id)
 {
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 9024e9ab52..32f3a675fa 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -270,9 +270,10 @@ static Port *create_port(char *name,
 			 Eterm pid,
 			 int *enop)
 {
+    ErtsPortTaskBusyPortQ *busy_port_queue;
     Port *prt;
     char *p;
-    size_t port_size, size;
+    size_t port_size, busy_port_queue_size, size;
     erts_aint32_t state = ERTS_PORT_SFLG_CONNECTED;
     erts_aint32_t x_pts_flgs = 0;
 #ifdef DEBUG
@@ -288,7 +289,13 @@ static Port *create_port(char *name,
     }
     else
 #endif
-	port_size = size = sizeof(Port);
+	port_size = size = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(Port));
+
+    busy_port_queue_size
+	= ((driver->flags & ERL_DRV_FLAG_NO_BUSY_MSGQ)
+	   ? 0
+	   : ERTS_ALC_DATA_ALIGN_SIZE(sizeof(ErtsPortTaskBusyPortQ)));
+    size += busy_port_queue_size;
 
     size += sys_strlen(name) + 1;
 
@@ -302,6 +309,13 @@ static Port *create_port(char *name,
     prt = (Port *) p;
     p += port_size;
 
+    if (!busy_port_queue_size)
+	busy_port_queue = NULL;
+    else {
+	busy_port_queue = (ErtsPortTaskBusyPortQ *) p;
+	p += busy_port_queue_size;
+    }
+
 #ifdef ERTS_SMP
     if (driver_lock) {
 	prt->lock = driver_lock;
@@ -320,6 +334,8 @@ static Port *create_port(char *name,
     prt->cleanup = 0;
 #endif
     
+    erts_port_task_pre_init_sched(&prt->sched, busy_port_queue);
+
     prt->name = p;
     sys_strcpy(p, name);
     prt->drv_ptr = driver;
@@ -503,8 +519,7 @@ erts_save_suspend_process_on_port(Port *prt, Process *process)
     erts_aint32_t flags;
     erts_port_task_sched_lock(&prt->sched);
     flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
-    saved = ((flags & (ERTS_PTS_FLG_BUSY
-		       | ERTS_PTS_FLG_EXIT)) == ERTS_PTS_FLG_BUSY);
+    saved = (flags & ERTS_PTS_FLGS_BUSY) && !(flags & ERTS_PTS_FLG_EXIT);
     if (saved)
 	erts_proclist_store_last(&prt->suspended, erts_proclist_create(process));
     erts_port_task_sched_unlock(&prt->sched);
@@ -1530,7 +1545,7 @@ bad_port_signal(Process *c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = flags;
+    sigdp->flags = (flags & ~ERTS_P2P_SIG_TYPE_MASK) | ERTS_P2P_SIG_TYPE_BAD;
 
     return erts_schedule_proc2port_signal(c_p,
 					  prt,
@@ -1736,7 +1751,7 @@ erts_port_output(Process *c_p,
     erts_driver_t *drv = prt->drv_ptr;
     size_t size;
     int try_call;
-    erts_aint32_t sched_flags, busy_flag, invalid_flags;
+    erts_aint32_t sched_flags, busy_flgs, invalid_flags;
     int task_flags;
     ErtsProc2PortSigCallback port_sig_callback;
     ErlDrvBinary *cbin = NULL;
@@ -1747,10 +1762,10 @@ erts_port_output(Process *c_p,
 		      | ERTS_PORT_SIG_FLG_NOSUSPEND
 		      | ERTS_PORT_SIG_FLG_FORCE)) == 0);
 
-    busy_flag = ((flags & ERTS_PORT_SIG_FLG_FORCE)
+    busy_flgs = ((flags & ERTS_PORT_SIG_FLG_FORCE)
 		 ? ((erts_aint32_t) 0)
-		 : ERTS_PTS_FLG_BUSY);
-    invalid_flags = busy_flag;
+		 : ERTS_PTS_FLGS_BUSY);
+    invalid_flags = busy_flgs;
     if (!refp)
 	invalid_flags |= ERTS_PTS_FLG_PARALLELISM;
 
@@ -1759,10 +1774,10 @@ erts_port_output(Process *c_p,
      */
 
     sched_flags = erts_smp_atomic32_read_nob(&prt->sched.flags);
-    if (sched_flags & (busy_flag|ERTS_PTS_FLG_EXIT))
-	return ((sched_flags & busy_flag)
-		? ERTS_PORT_OP_BUSY
-		: ERTS_PORT_OP_DROPPED);
+    if (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))
+	return ((sched_flags & ERTS_PTS_FLG_EXIT)
+		? ERTS_PORT_OP_DROPPED
+		: ERTS_PORT_OP_BUSY);
 
     try_call = !(sched_flags & (invalid_flags|ERTS_PTS_FLGS_FORCE_SCHEDULE_OP));
 
@@ -1949,6 +1964,7 @@ erts_port_output(Process *c_p,
 	}
 
 	sigdp = erts_port_task_alloc_p2p_sig_data();
+	sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUTV;
 	sigdp->u.outputv.from = from;
 	sigdp->u.outputv.evp = evp;
 	sigdp->u.outputv.cbinp = cbin;
@@ -2038,6 +2054,7 @@ erts_port_output(Process *c_p,
 	}
 
 	sigdp = erts_port_task_alloc_p2p_sig_data();
+	sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUT;
 	sigdp->u.output.from = from;
 	sigdp->u.output.bufp = buf;
 	sigdp->u.output.size = size;
@@ -2045,7 +2062,7 @@ erts_port_output(Process *c_p,
     }
 
     task_flags = ERTS_PT_FLG_WAIT_BUSY;
-    sigdp->flags = flags;
+    sigdp->flags |= flags;
     if (flags & (ERTS_P2P_SIG_DATA_FLG_FORCE|ERTS_P2P_SIG_DATA_FLG_NOSUSPEND)) {
 	task_flags = 0;
 	if (flags & ERTS_P2P_SIG_DATA_FLG_FORCE)
@@ -2070,7 +2087,7 @@ erts_port_output(Process *c_p,
 	return res;
     }
 
-    if ((sched_flags & (busy_flag|ERTS_PTS_FLG_EXIT)) == ERTS_PTS_FLG_BUSY)
+    if (!(sched_flags & ERTS_PTS_FLG_EXIT) && (sched_flags & busy_flgs))
 	return ERTS_PORT_OP_BUSY_SCHEDULED;
 
     return res;
@@ -2201,7 +2218,7 @@ erts_port_exit(Process *c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = flags;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_EXIT | flags;
     sigdp->u.exit.from = from;
 
     if (is_immed(reason)) {
@@ -2367,7 +2384,7 @@ erts_port_connect(Process *c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = flags;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_CONNECT | flags;
 
     sigdp->u.connect.from = from;
     sigdp->u.connect.connected = connect_id;
@@ -2424,7 +2441,7 @@ erts_port_unlink(Process *c_p, Port *prt, Eterm from, Eterm *refp)
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = 0;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_UNLINK;
     sigdp->u.unlink.from = from;
     
     return erts_schedule_proc2port_signal(c_p,
@@ -2509,7 +2526,7 @@ erts_port_link(Process *c_p, Port *prt, Eterm to, Eterm *refp)
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = 0;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_LINK;
     sigdp->u.link.port = prt->common.id;
     sigdp->u.link.to = to;
     
@@ -3851,7 +3868,7 @@ erts_port_control(Process* c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = 0;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_CONTROL;
     sigdp->u.control.binp = binp;
     sigdp->u.control.command = command;
     sigdp->u.control.bufp = bufp;
@@ -4131,7 +4148,7 @@ erts_port_call(Process* c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = 0;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_CALL;
     sigdp->u.call.command = command;
     sigdp->u.call.bufp = bufp;
     sigdp->u.call.size = size;
@@ -4298,7 +4315,7 @@ erts_port_info(Process* c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = 0;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_INFO;
     sigdp->u.info.item = item;
 
     return erts_schedule_proc2port_signal(c_p,
@@ -4383,7 +4400,7 @@ erts_port_set_data(Process* c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = 0;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_SET_DATA;
     sigdp->u.set_data.data = set_data;
     sigdp->u.set_data.bp = bp;
 
@@ -4516,7 +4533,7 @@ erts_port_get_data(Process* c_p,
     }
 
     sigdp = erts_port_task_alloc_p2p_sig_data();
-    sigdp->flags = 0;
+    sigdp->flags = ERTS_P2P_SIG_TYPE_GET_DATA;
 
     return erts_schedule_proc2port_signal(c_p,
 					  prt,
@@ -4608,8 +4625,8 @@ set_busy_port(ErlDrvPort dprt, int on)
 
     if (on) {
 	flags = erts_smp_atomic32_read_bor_acqb(&prt->sched.flags,
-						ERTS_PTS_FLG_BUSY);
-	if (flags & ERTS_PTS_FLG_BUSY)
+						ERTS_PTS_FLG_BUSY_PORT);
+	if (flags & ERTS_PTS_FLG_BUSY_PORT)
 	    return; /* Already busy */
 
 	if (flags & ERTS_PTS_FLG_HAVE_NS_TASKS)
@@ -4623,11 +4640,9 @@ set_busy_port(ErlDrvPort dprt, int on)
         }
 #endif
     } else {
-        ErtsProcList *plp;
-
 	flags = erts_smp_atomic32_read_band_acqb(&prt->sched.flags,
-						 ~ERTS_PTS_FLG_BUSY);
-	if (!(flags & ERTS_PTS_FLG_BUSY))
+						 ~ERTS_PTS_FLG_BUSY_PORT);
+	if (!(flags & ERTS_PTS_FLG_BUSY_PORT))
 	    return; /* Already non-busy */
 
 #ifdef USE_VM_PROBES
@@ -4645,52 +4660,60 @@ set_busy_port(ErlDrvPort dprt, int on)
 	    erts_dist_port_not_busy(prt);
 	}
 
-	/*
-	 * Resume, in a round-robin fashion, all processes waiting on the port.
-	 * 
-	 * This version submitted by Tony Rogvall. The earlier version used
-	 * to resume the processes in order, which caused starvation of all but
-	 * the first process.
-	 */
+	if (!(flags & ERTS_PTS_FLG_BUSY_PORT_Q))
+	    erts_port_resume_procs(prt);
+    }
+}
 
-	erts_port_task_sched_lock(&prt->sched);
-	plp = prt->suspended;
-	prt->suspended = NULL;
-	erts_port_task_sched_unlock(&prt->sched);
+void
+erts_port_resume_procs(Port *prt)
+{
+    /*
+     * Resume, in a round-robin fashion, all processes waiting on the port.
+     * 
+     * This version submitted by Tony Rogvall. The earlier version used
+     * to resume the processes in order, which caused starvation of all but
+     * the first process.
+     */
+    ErtsProcList *plp;
 
-        if (erts_proclist_fetch(&plp, NULL)) {
+    erts_port_task_sched_lock(&prt->sched);
+    plp = prt->suspended;
+    prt->suspended = NULL;
+    erts_port_task_sched_unlock(&prt->sched);
+
+    if (erts_proclist_fetch(&plp, NULL)) {
 
 #ifdef USE_VM_PROBES
-            /*
-             * Hrm, for blocked dist ports, plp always seems to be NULL.
-             * That's not so fun.
-             * Well, another way to get the same info is using a D
-             * script to correlate an earlier process-port_blocked+pid
-             * event with a later process-scheduled event.  That's
-             * subject to the multi-CPU races with how events are
-             * handled, but hey, that way works most of the time.
-             */
-            if (DTRACE_ENABLED(process_port_unblocked)) {
-                DTRACE_CHARBUF(pid_str, 16);
-                ErtsProcList* plp2 = plp;
-
-                erts_snprintf(port_str, sizeof(port_str),
-                             "%T", prt->common.id);
-                while (plp2 != NULL) {
-                    erts_snprintf(pid_str, sizeof(pid_str), "%T", plp2->pid);
-                    DTRACE2(process_port_unblocked, pid_str, port_str);
-                }
-            }
-#endif
+	/*
+	 * Hrm, for blocked dist ports, plp always seems to be NULL.
+	 * That's not so fun.
+	 * Well, another way to get the same info is using a D
+	 * script to correlate an earlier process-port_blocked+pid
+	 * event with a later process-scheduled event.  That's
+	 * subject to the multi-CPU races with how events are
+	 * handled, but hey, that way works most of the time.
+	 */
+	if (DTRACE_ENABLED(process_port_unblocked)) {
+	    DTRACE_CHARBUF(port_str, 16);
+	    DTRACE_CHARBUF(pid_str, 16);
+	    ErtsProcList* plp2 = plp;
 
-            /* First proc should be resumed last */
-	    if (plp->next) {
-		plp->next->prev = NULL;
-		erts_resume_processes(plp->next);
-		plp->next = NULL;
+	    erts_snprintf(port_str, sizeof(port_str), "%T", prt->common.id);
+	    while (plp2 != NULL) {
+		erts_snprintf(pid_str, sizeof(pid_str), "%T", plp2->pid);
+		DTRACE2(process_port_unblocked, pid_str, port_str);
 	    }
-	    erts_resume_processes(plp);
-        }
+	}
+#endif
+
+	/* First proc should be resumed last */
+	if (plp->next) {
+	    plp->next->prev = NULL;
+	    erts_resume_processes(plp->next);
+	    plp->next = NULL;
+	}
+	erts_resume_processes(plp);
     }
 }
 
diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c
index 4a87b0a385..f68ce68407 100644
--- a/erts/emulator/drivers/common/inet_drv.c
+++ b/erts/emulator/drivers/common/inet_drv.c
@@ -692,6 +692,8 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n)
 #define INET_LOPT_UDP_READ_PACKETS 33  /* Number of packets to read */
 #define INET_OPT_RAW               34  /* Raw socket options */
 #define INET_LOPT_TCP_SEND_TIMEOUT_CLOSE 35  /* auto-close on send timeout or not */
+#define INET_LOPT_TCP_MSGQ_HIWTRMRK     36  /* set local high watermark */
+#define INET_LOPT_TCP_MSGQ_LOWTRMRK     37  /* set local low watermark */
 /* SCTP options: a separate range, from 100: */
 #define SCTP_OPT_RTOINFO		100
 #define SCTP_OPT_ASSOCINFO		101
@@ -808,6 +810,8 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n)
 
 #define INET_HIGH_WATERMARK (1024*8) /* 8k pending high => busy  */
 #define INET_LOW_WATERMARK  (1024*4) /* 4k pending => allow more */
+#define INET_HIGH_MSGQ_WATERMARK (1024*8) /* 8k pending high => busy  */
+#define INET_LOW_MSGQ_WATERMARK  (1024*4) /* 4k pending => allow more */
 
 #define INET_INFINITY  0xffffffff  /* infinity value */
 
@@ -5537,6 +5541,28 @@ static int inet_set_opts(inet_descriptor* desc, char* ptr, int len)
 	    }
 	    continue;
 
+	case INET_LOPT_TCP_MSGQ_HIWTRMRK:
+	    if (desc->stype == SOCK_STREAM) {
+		ErlDrvSizeT high;
+		if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
+		    || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
+		    return -1;
+		high = (ErlDrvSizeT) ival;
+		erl_drv_busy_msgq_limits(desc->port, NULL, &high);
+	    }
+	    continue;
+
+	case INET_LOPT_TCP_MSGQ_LOWTRMRK:
+	    if (desc->stype == SOCK_STREAM) {
+		ErlDrvSizeT low;
+		if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
+		    || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
+		    return -1;
+		low = (ErlDrvSizeT) ival;
+		erl_drv_busy_msgq_limits(desc->port, &low, NULL);
+	    }
+	    continue;
+
 	case INET_LOPT_TCP_SEND_TIMEOUT:
 	    if (desc->stype == SOCK_STREAM) {
 		tcp_descriptor* tdesc = (tcp_descriptor*) desc;
@@ -6422,6 +6448,32 @@ static ErlDrvSSizeT inet_fill_opts(inet_descriptor* desc,
 	    }
 	    continue;
 
+	case INET_LOPT_TCP_MSGQ_HIWTRMRK:
+	    if (desc->stype == SOCK_STREAM) {
+		ErlDrvSizeT high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+		*ptr++ = opt;
+		erl_drv_busy_msgq_limits(desc->port, NULL, &high);
+		ival = high > INT_MAX ? INT_MAX : (int) high;
+		put_int32(ival, ptr);
+	    }
+	    else {
+		TRUNCATE_TO(0,ptr);
+	    }
+	    continue;
+
+	case INET_LOPT_TCP_MSGQ_LOWTRMRK:
+	    if (desc->stype == SOCK_STREAM) {
+		ErlDrvSizeT low = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+		*ptr++ = opt;
+		erl_drv_busy_msgq_limits(desc->port, &low, NULL);
+		ival = low > INT_MAX ? INT_MAX : (int) low;
+		put_int32(ival, ptr);
+	    }
+	    else {
+		TRUNCATE_TO(0,ptr);
+	    }
+	    continue;
+
 	case INET_LOPT_TCP_SEND_TIMEOUT:
 	    if (desc->stype == SOCK_STREAM) {
 		*ptr++ = opt;
@@ -8029,6 +8081,7 @@ static int tcp_inet_init(void)
 
 static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
 {
+    ErlDrvSizeT q_low, q_high;
     tcp_descriptor* desc;
     DEBUGF(("tcp_inet_start(%ld) {\r\n", (long)port));
 
@@ -8038,6 +8091,17 @@ static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
 	return ERL_DRV_ERROR_ERRNO;
     desc->high = INET_HIGH_WATERMARK;
     desc->low  = INET_LOW_WATERMARK;
+    q_high = INET_HIGH_MSGQ_WATERMARK;
+    q_low = INET_LOW_MSGQ_WATERMARK;
+    if (q_low < ERL_DRV_BUSY_MSGQ_LIM_MIN)
+	q_low = ERL_DRV_BUSY_MSGQ_LIM_MIN;
+    else if (q_low > ERL_DRV_BUSY_MSGQ_LIM_MAX)
+	q_low = ERL_DRV_BUSY_MSGQ_LIM_MAX;
+    if (q_high < ERL_DRV_BUSY_MSGQ_LIM_MIN)
+	q_high = ERL_DRV_BUSY_MSGQ_LIM_MIN;
+    else if (q_high > ERL_DRV_BUSY_MSGQ_LIM_MAX)
+	q_high = ERL_DRV_BUSY_MSGQ_LIM_MAX;
+    erl_drv_busy_msgq_limits(port, &q_low, &q_high);
     desc->send_timeout = INET_INFINITY;
     desc->send_timeout_close = 0;
     desc->busy_on_send = 0;
@@ -8061,6 +8125,7 @@ static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
 static tcp_descriptor* tcp_inet_copy(tcp_descriptor* desc,SOCKET s,
 				     ErlDrvTermData owner, int* err)
 {
+    ErlDrvSizeT q_low, q_high;
     ErlDrvPort port = desc->inet.port;
     tcp_descriptor* copy_desc;
 
@@ -8099,6 +8164,13 @@ static tcp_descriptor* tcp_inet_copy(tcp_descriptor* desc,SOCKET s,
 	FREE(copy_desc);
 	return NULL;
     }
+
+    /* Read busy msgq limits of parent */
+    q_low = q_high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+    erl_drv_busy_msgq_limits(desc->inet.port, &q_low, &q_high);
+    /* Write same busy msgq limits to child */
+    erl_drv_busy_msgq_limits(port, &q_low, &q_high);
+
     copy_desc->inet.port = port;
     copy_desc->inet.dport = driver_mk_port(port);
     *err = 0;
diff --git a/erts/preloaded/ebin/prim_inet.beam b/erts/preloaded/ebin/prim_inet.beam
index ae9c17bbcb..e5c1de86f4 100644
--- a/erts/preloaded/ebin/prim_inet.beam
+++ b/erts/preloaded/ebin/prim_inet.beam
diff --git a/erts/preloaded/src/prim_inet.erl b/erts/preloaded/src/prim_inet.erl
index 846ae97ed2..f7a9718cde 100644
--- a/erts/preloaded/src/prim_inet.erl
+++ b/erts/preloaded/src/prim_inet.erl
@@ -1071,6 +1071,8 @@ enc_opt(deliver)         -> ?INET_LOPT_DELIVER;
 enc_opt(exit_on_close)   -> ?INET_LOPT_EXITONCLOSE;
 enc_opt(high_watermark)  -> ?INET_LOPT_TCP_HIWTRMRK;
 enc_opt(low_watermark)   -> ?INET_LOPT_TCP_LOWTRMRK;
+enc_opt(high_msgq_watermark)  -> ?INET_LOPT_TCP_MSGQ_HIWTRMRK;
+enc_opt(low_msgq_watermark)   -> ?INET_LOPT_TCP_MSGQ_LOWTRMRK;
 enc_opt(bit8)            -> ?INET_LOPT_BIT8;
 enc_opt(send_timeout)    -> ?INET_LOPT_TCP_SEND_TIMEOUT;
 enc_opt(send_timeout_close) -> ?INET_LOPT_TCP_SEND_TIMEOUT_CLOSE;
@@ -1125,6 +1127,8 @@ dec_opt(?INET_LOPT_DELIVER)       -> deliver;
 dec_opt(?INET_LOPT_EXITONCLOSE)   -> exit_on_close;
 dec_opt(?INET_LOPT_TCP_HIWTRMRK)  -> high_watermark;
 dec_opt(?INET_LOPT_TCP_LOWTRMRK)  -> low_watermark;
+dec_opt(?INET_LOPT_TCP_MSGQ_HIWTRMRK)  -> high_msgq_watermark;
+dec_opt(?INET_LOPT_TCP_MSGQ_LOWTRMRK)  -> low_msgq_watermark;
 dec_opt(?INET_LOPT_BIT8)          -> bit8;
 dec_opt(?INET_LOPT_TCP_SEND_TIMEOUT) -> send_timeout;
 dec_opt(?INET_LOPT_TCP_SEND_TIMEOUT_CLOSE) -> send_timeout_close;
@@ -1220,6 +1224,8 @@ type_opt_1(deliver) ->
 type_opt_1(exit_on_close)   -> bool;
 type_opt_1(low_watermark)   -> int;
 type_opt_1(high_watermark)  -> int;
+type_opt_1(low_msgq_watermark)   -> int;
+type_opt_1(high_msgq_watermark)  -> int;
 type_opt_1(bit8) ->
     {enum,[{clear, ?INET_BIT8_CLEAR},
 	   {set,   ?INET_BIT8_SET},
diff --git a/lib/kernel/doc/src/inet.xml b/lib/kernel/doc/src/inet.xml
index bf6c4cfb1a..bfdb163e17 100644
--- a/lib/kernel/doc/src/inet.xml
+++ b/lib/kernel/doc/src/inet.xml
@@ -487,6 +487,36 @@ fe80::204:acff:fe17:bf38
               example <c>Size == 2</c>, the data received will match
               <c>[Byte1,Byte2|Binary]</c>.</p>
           </item>
+          <tag><c>{high_msgq_watermark, Size}</c> (TCP/IP sockets)</tag>
+          <item>
+	    <p>The socket message queue will be set into a busy
+	    state when the amount of data queued on the message
+	    queue reaches this limit. Note that this limit only
+	    concerns data that have not yet reached the ERTS internal
+	    socket implementation. Default value used is 8 kB.</p>
+	    <p>Senders of data to the socket will be suspended if
+	    either the socket message queue is busy, or the socket
+	    itself is busy.</p>
+	    <p>For more information see the <c>low_msgq_watermark</c>,
+	    <c>high_watermark</c>, and <c>low_watermark</c> options.</p>
+	    <p>Note that distribution sockets will disable the use of
+	    <c>high_msgq_watermark</c> and <c>low_msgq_watermark</c>,
+	    and will instead use the
+	    <seealso marker="erts:erlang#system_info_dist_buf_busy_limit">distribution
+	    buffer busy limit</seealso> which is a similar feature.</p>
+          </item>
+          <tag><c>{high_watermark, Size}</c> (TCP/IP sockets)</tag>
+          <item>
+	    <p>The socket will be set into a busy state when the amount
+	    of data queued internally by the ERTS socket implementation
+	    reaches this limit. Default value used is 8 kB.</p>
+	    <p>Senders of data to the socket will be suspended if
+	    either the socket message queue is busy, or the socket
+	    itself is busy.</p>
+	    <p>For more information see the <c>low_watermark</c>,
+	    <c>high_msgq_watermark</c>, and <c>low_msqg_watermark</c>
+	    options.</p>
+          </item>
           <tag><c>{keepalive, Boolean}</c>(TCP/IP sockets)</tag>
           <item>
             <p>Enables/disables periodic transmission on a connected
@@ -495,6 +525,40 @@ fe80::204:acff:fe17:bf38
               considered broken and an error message will be sent to
               the controlling process. Default disabled.</p>
           </item>
+          <tag><c>{low_msgq_watermark, Size}</c> (TCP/IP sockets)</tag>
+          <item>
+	    <p>If the socket message queue is in a busy state, the
+	    socket message queue will be set in a not busy state when
+	    the amount of data queued in the message queue falls
+	    below this limit. Note that this limit only concerns data
+	    that have not yet reached the ERTS internal socket
+	    implementation. Default value used is 4 kB.</p>
+	    <p>Senders that have been suspended due to either a
+	    busy message queue or a busy socket, will be resumed
+	    when neither the socket message queue, nor the socket
+	    are busy.</p>
+	    <p>For more information see the <c>high_msgq_watermark</c>,
+	    <c>high_watermark</c>, and <c>low_watermark</c> options.</p>
+	    <p>Note that distribution sockets will disable the use of
+	    <c>high_msgq_watermark</c> and <c>low_msgq_watermark</c>,
+	    and will instead use the
+	    <seealso marker="erts:erlang#system_info_dist_buf_busy_limit">distribution
+	    buffer busy limit</seealso> which is a similar feature.</p>
+          </item>
+          <tag><c>{low_watermark, Size}</c> (TCP/IP sockets)</tag>
+          <item>
+	    <p>If the socket is in a busy state, the socket will
+	    be set in a not busy state when the amount of data
+	    queued internally by the ERTS socket implementation
+	    falls below this limit. Default value used is 4 kB.</p>
+	    <p>Senders that have been suspended due to either a
+	    busy message queue or a busy socket, will be resumed
+	    when neither the socket message queue, nor the socket
+	    are busy.</p>
+	    <p>For more information see the <c>high_watermark</c>,
+	    <c>high_msgq_watermark</c>, and <c>low_msgq_watermark</c>
+	    options.</p>
+          </item>
           <tag><c>{nodelay, Boolean}</c>(TCP/IP sockets)</tag>
           <item>
             <p>If <c>Boolean == true</c>, the <c>TCP_NODELAY</c> option
diff --git a/lib/kernel/src/gen_tcp.erl b/lib/kernel/src/gen_tcp.erl
index ef6bfdf7f4..519c1bb8b4 100644
--- a/lib/kernel/src/gen_tcp.erl
+++ b/lib/kernel/src/gen_tcp.erl
@@ -38,9 +38,11 @@
         {dontroute,       boolean()} |
         {exit_on_close,   boolean()} |
         {header,          non_neg_integer()} |
+        {high_msgq_watermark, pos_integer()} |
         {high_watermark,  non_neg_integer()} |
         {keepalive,       boolean()} |
         {linger,          {boolean(), non_neg_integer()}} |
+        {low_msgq_watermark, pos_integer()} |
         {low_watermark,   non_neg_integer()} |
         {mode,            list | binary} | list | binary |
         {nodelay,         boolean()} |
@@ -68,9 +70,11 @@
         dontroute |
         exit_on_close |
         header |
+        high_msgq_watermark |
         high_watermark |
         keepalive |
         linger |
+        low_msgq_watermark |
         low_watermark |
         mode |
         nodelay |
diff --git a/lib/kernel/src/inet.erl b/lib/kernel/src/inet.erl
index abaf4486dc..e791bb135f 100644
--- a/lib/kernel/src/inet.erl
+++ b/lib/kernel/src/inet.erl
@@ -535,6 +535,7 @@ options() ->
      buffer, header, active, packet, deliver, mode,
      multicast_if, multicast_ttl, multicast_loop,
      exit_on_close, high_watermark, low_watermark,
+     high_msgq_watermark, low_msgq_watermark,
      bit8, send_timeout, send_timeout_close
     ].
 
@@ -552,8 +553,9 @@ stats() ->
 connect_options() ->
     [tos, priority, reuseaddr, keepalive, linger, sndbuf, recbuf, nodelay,
      header, active, packet, packet_size, buffer, mode, deliver,
-     exit_on_close, high_watermark, low_watermark, bit8, send_timeout,
-     send_timeout_close, delay_send,raw].
+     exit_on_close, high_watermark, low_watermark, high_msgq_watermark,
+     low_msgq_watermark, bit8, send_timeout, send_timeout_close,
+     delay_send,raw].
     
 connect_options(Opts, Family) ->
     BaseOpts = 
@@ -608,8 +610,9 @@ con_add(Name, Val, R, Opts, AllOpts) ->
 listen_options() ->
     [tos, priority, reuseaddr, keepalive, linger, sndbuf, recbuf, nodelay,
      header, active, packet, buffer, mode, deliver, backlog,
-     exit_on_close, high_watermark, low_watermark, bit8, send_timeout,
-     send_timeout_close, delay_send, packet_size,raw].
+     exit_on_close, high_watermark, low_watermark, high_msgq_watermark,
+     low_msgq_watermark, bit8, send_timeout, send_timeout_close,
+     delay_send, packet_size,raw].
 
 listen_options(Opts, Family) ->
     BaseOpts = 
diff --git a/lib/kernel/src/inet_int.hrl b/lib/kernel/src/inet_int.hrl
index cf893c73eb..2dfbf75d25 100644
--- a/lib/kernel/src/inet_int.hrl
+++ b/lib/kernel/src/inet_int.hrl
@@ -141,6 +141,8 @@
 -define(INET_LOPT_READ_PACKETS,  33).
 -define(INET_OPT_RAW,            34).
 -define(INET_LOPT_TCP_SEND_TIMEOUT_CLOSE, 35).
+-define(INET_LOPT_TCP_MSGQ_HIWTRMRK,  36).
+-define(INET_LOPT_TCP_MSGQ_LOWTRMRK,  37).
 % Specific SCTP options: separate range:
 -define(SCTP_OPT_RTOINFO,	 	100).
 -define(SCTP_OPT_ASSOCINFO,	 	101).