. . .
     callback => {
          data => \&data,
          add  => \&start_new_cnx,
     },
 . . .

# Let the connection start up, and spawn a thread to handle it.
# Returns 1 for both parent and child task, so the connection isn't closed.
sub start_new_cnx {
   $pm->start;
   return 1;
}

sub data {
   # the same as you had it, but get rid of the $pm->start call, as it's
   # already been started.  Just call $pm->finish when you want to end
   # the connection and terminate the task.
}