Re: sharing objects

Christopher Fowler wrote:
> On Tue, 2006-10-10 at 09:23 -0500, Michael J. Pomraning wrote:
>> Perhaps pass an unblest reference through perl's prefab Thread::Queue
>> class?
> 
> That should be the best way.  Is there any great way to share objects
> between threads?
> 
> 

Sharing objects between threads is simple: just make sure that you
use a shared array or hash for your object, and only assign either
simple scalars or shared refs to its members. Then you can pass
the objects between threads via a simple Thread::Queue object.

I've done a quick rewrite of your example (note I have no idea what
the function of your script is, but the Packet class is a pretty simple container
class, so hopefully this captures what you're attempting).

Some major issues I've noticed:

1) using a `host $addr` to do resolution may be a problem, since
you're shelling out from a thread...I'm not certain how well that's going to work.
I've noted the use of gethostbyaddr() as an alternate (tho I don't know how thread
safe it is either)

2) You never waited to join() your threads; I don't know what effect that
might have on your exit condition

3) The way you setup your object sharing was a bit obtuse; I've simplified it

4) I've replaced your @obj ad-hoc queue with Thread::Queue, which handles
the needed locking  (which it appears you weren't applying)

Code below. NOTE: I've not run it, it likely has some syntax errors.

HTH,
Dean Arnold
Presicient Corp.


package Packet;

use threads;
use threads::shared;
use strict;
use warnings;

sub new {
	my ($class, $v1, $v2) = @_;
#
#	NOTE: this assume that the packet and timestamp
#	elements are simple scalars; if not, then you'll
#	need more share()'ing to coerce them into something
#	that can be assigned to a shared hash
#
	my %pkt : shared = (
		packet => $v1,
		timestamp => $v2
	);
	return bless \%pkt, $class;
}

sub get_timestamp {
	my $self = shift;
#
#	FYI/Warning: I've encountered some instances where
#	derefencing an element off a shared hash would
#	silently discard the returned value, and had to do
#	an intermediate assignment to a lexical variable
#	before using the value; I've not yet determined
#	the cause of the issue yet
#
	return $self->{'timestamp'};
}

sub get_packet {
	my $self = shift;
#
#	see caveat above
#
	return $self->{'packet'};
}

1;

package main;

use Net::PcapUtils;
use NetPacket::Ethernet;
use NetPacket::IP;
use NetPacket::TCP;
use threads;
use threads::shared;
use Thread::Queue;
use Data::Dumper;

use strict;
use warnings;

$| = 1;
my $prog = "tcp[13] = 2 and src net not 192.168.2";
#
#	create a thread queue
#
my $q = Thread::Queue->new();

sub get_timestamp {
	my $ti = shift;
	$ti = time() unless $ti;
	my ($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst) =
		localtime($ti);

	my $text = sprintf("%02d-%02d %02d:%02d:%02d",
		($mon+1), $mday, $hour, $min, $sec);
	return $text;

}

sub now {
	return time();
}

sub resolv {
	my $addr = shift;
#
#	OUCH! this may never work properly...forking from a thread
#	may lead to chaos, esp. on Win32...isn't there a better
#	Perlish solution (e.g., get host by addr ?
#
#	my $host = gethostbyaddr($addr, AF_INET);
#
	my $host = `host $addr`;
	chomp $host;

	return $addr if $host =~ m/not found/;
	return $addr unless $host =~ m/pointer\s(.+?)\.$/;

	$addr = $1;
	return $addr;

}

sub process_thread {
	while(1) {

#
#	wait for a packet (or for 'DIE')
#
		my $pktobj = $q->dequeue();
#
#	if not a ref, then it must be 'DIE'
#
		return 1
			unless ref $pktobj;
#
# Get the packet off the list and unpack the time
#
		my $pkt = $pktobj->get_packet();
		my $ti = $pktobj->get_timestamp();

# get the source IP adrress
		my $src_ip = NetPacket::IP->decode(
			NetPacket::Ethernet::strip($pkt))->{src_ip};

# I want incoming
		next if $src_ip =~ m/209\.168\.246\.233/;

		my $pt = NetPacket::TCP->decode(
			 NetPacket::IP::strip(
				 NetPacket::Ethernet::strip($pkt)));

		print get_timestamp($ti)." SYN ".resolv($src_ip)."($src_ip) -> $pt->{'dest_port'}\n";
	}
}

sub catch {
	my ($arg,$hdr,$pkt) = @_ ;
#
#	enqueue the packet for processing
#
	$q->enqueue(Packet->new($pkt, now()));
}

#
#	spawn the thread
#
my $pthr = threads->new(\&process_thread);
#
#	start snooping
#
Net::PcapUtils::loop(\&catch, FILTER=> $prog);
#
#	I don't know what your exit condition is ???
#	but here's how I've instrumented cleanup
#	NOTE that if you start >1 thread, you'll need
#	to send multiple DIE's, and join all the threads
#
$q->enqueue('DIE');
$pthr->join();